{
  "param1": "{\"job.name\":\"简化版本-CheckPoint-实时归集企业人员信息-3.0\",\"checkpoint.mode\":\"file\",\"checkpoint.hdfs.config\":\"core-site-default.xml\",\"checkpoint.disk.location\":\"/tmp/cp-table-company\"}",
  "param2": "{\n  \"name\": \"实时归集企业-信息\",\n  \"jdbcMeta\": {\n    \"url\": \"jdbc:mysql://192.168.0.7:3368/dispatch?characterEncoding=UTF-8&connectionTimeZone=GMT%2B8&allowMultiQueries=true&autoReconnect=true\",\n    \"userName\": \"v5_enterprise2021\",\n    \"password\": \"Prod_v5#202109\",\n    \"sqlMetaDefinitions\": [\n      {\n        \"tableName\": \"rd_company_summary\",\n        \"sql\": \"create table jdbc_insert_rd_company_summary(id STRING,status INT, name STRING,legal STRING,legal_telephone STRING,institutional STRING,register_address STRING,business_address STRING,service_scope STRING,lon_lat STRING,organization_id STRING,supervise_domain_id STRING,supervise_depart_id STRING, last_sync_time TIMESTAMP(3),last_updated_time TIMESTAMP(3),supervise_region_code STRING,PRIMARY KEY (id) NOT ENFORCED)\"\n      }\n    ]\n  },\n  \"kafkaMeta\": {\n    \"autoOffsetRest\": \"earliest\",\n    \"scanStartupMode\": \"earliest-offset\",\n    \"enableAutoCommit\": \"true\",\n    \"bootstrapServer\": \"192.168.0.52:9092\",\n    \"topicDefinitions\": [\n      {\n        \"topic\": \"obpm2.binlog-cdc.topic.data.tenant_organizations\",\n        \"sql\": \"create table kafka_tenant_organizations(id STRING,name STRING,industry_code STRING,institutional_code STRING,place_of_business_address STRING,place_of_register_address STRING,latitude STRING,longitude STRING,PRIMARY KEY (id) NOT ENFORCED) \"\n      },\n      {\n        \"topic\": \"baibaodunflow.binlog-cdc.topic.data.tlk_companyinformation\",\n        \"sql\": \"create table kafka_tlk_companyinformation(ID STRING,ITEM_companyStatus STRING, DOMAINID STRING, ITEM_legalPerson STRING,ITEM_legalPersonPhone STRING,ITEM_businessScope STRING,LASTMODIFIED BIGINT,PRIMARY KEY (ID) NOT ENFORCED)  \"\n      }\n    ]\n  },\n  \"executeSql\": [\n    \"insert into jdbc_insert_rd_company_summary(id,name,status,legal,legal_telephone,institutional,register_address, business_address,service_scope,organization_id,supervise_domain_id,supervise_depart_id,last_sync_time,last_updated_time,supervise_region_code,lon_lat)select distinct o.id,o.`name`,(case when ITEM_companyStatus='注销' then 2 when ITEM_companyStatus='撤销' then 1 else 0 end) as ITEM_companyStatus,ci.ITEM_legalPerson,ci.ITEM_legalPersonPhone,ifnull(o.institutional_code,''),o.place_of_register_address,o.place_of_business_address,ci.ITEM_businessScope,o.id,'#','#supervise_depart_id',TO_TIMESTAMP(FROM_UNIXTIME(ci.LASTMODIFIED/1000, 'yyyy-MM-dd HH:mm:ss')),TO_TIMESTAMP(FROM_UNIXTIME(ci.LASTMODIFIED/1000, 'yyyy-MM-dd HH:mm:ss')),'flink', convert2Point(o.longitude,o.latitude) as lonlat from kafka_tenant_organizations o left join kafka_tlk_companyinformation ci on o.id=ci.DOMAINID\"\n  ]\n}"
}