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'