insert into sink_flink_tenant_organizations
select o.id,o.created_time,o.name,
       (select k.ITEM_attendanceSiteName from source_kafka_tlk_attendance_site_base_info k
               where k.DOMAINID=o.id and k.ITEM_attendanceSiteState='生效'
               order by ITEM_attendanceStartDate desc limit 1) as place_of_business_address,
       (select count(distinct k.id) from source_kafka_tenant_employees k
               where k.organization_id=o.id and k.occupation_type=1 and k.status=0) as numOfSecurityMen,
        (select count(distinct k.ID) from source_kafka_tlk_attendance_site_base_info k
               where k.DOMAINID=o.id and k.ITEM_attendanceSiteState='生效') as numOfSecurityStation
from source_kafka_tenant_organizations o;

insert into sink_flink_tenant_organizations
select o.id,o.created_time,o.name,
        max(k.ITEM_attendanceSiteName) as place_of_business_address,
        count(distinct t.id) as numOfSecurityMen,
        count(distinct k.ID) as numOfSecurityStation
from source_kafka_tenant_organizations o
left join source_kafka_tlk_attendance_site_base_info k on o.id=k.DOMAINID and k.ITEM_attendanceSiteState='生效'
left join source_kafka_tenant_employees t on t.organization_id=o.id and t.occupation_type=1 and t.status=0
group by o.id,o.created_time,o.name