1 流程图

2 Flink来源表建模

--来源-城市topic
CREATE TABLE NJ_QL_JC_SSJC_SOURCE (
record string 
) WITH ('connector' = 'kafka','topic' = 'QL_JC_SSJC','properties.bootstrap.servers' = '172.*.*.*:9092','properties.group.id' = 'QL_JC_SSJC_NJ_QL_JC_SSJC_SOURCE','scan.startup.mode' = 'group-offsets','properties.isolation.level' = 'read_committed','properties.auto.offset.reset' = 'earliest','format' = 'raw'
);
--来源-中台kafka-topic
CREATE TABLE ODS_QL_JC_SSJC_SOURCE (
sscsdm string,
extract_time TIMESTAMP,
record string
) WITH ('connector' = 'kafka','topic' = 'ODS_QL_JC_SSJC','properties.bootstrap.servers'