读取Message
md_customer_divide_info
维表Join（1:n)
ck1
READY
Write
单表处理算子
ck3
运营实时计算方案
INSERT/REACT/UPSERTOutputFormat1
分解/去重
目标Table1
md_customer_synergy_info
创建table创建分区
实时同步维表延迟10s
ck4
不存在
提取biz_id
ReplacingMergeTreeOn ck1
Message
表/分区是否存在
ddl
OutputFormat2
ES 宽表md_customer_info
目标Table3
加入父OutputFormat Map
RUNNING
Replicate
ReplacingMergeTreeOn ckbackup
读到
Read
ckbackup
Shunter (大拆小)
Exception
Kafka 输出流
OutputFormat3
nginx SLB1
datarkWrite
DB
cc_group_extend_info
database1.table1
Message 流
制造延时流最大延迟40s
根据db，tablename从Map中查找子OutputFormat
维表Join(1:1)
dml
ReplicatedReplacingMergeTreeOn Cluster1234
binlog
表灾备方案
DistributedOn Cluster1b
子OutputFormatWriteRecord 写表
提取customer_id
不需停止
生产部署
读取错误
DDL Alert
基础生产部署
ck2
ReplicatedReplacingMergeTreeOn Cluster12
wirte错误？是否停止
开始
结束
需要停止
isDdl?
目标Table2
外部接口Read
Cluster1234
cc_group_profile_info
DistributedOn Cluster1234
实时同步实例状态转换图
Writer 流程图
表存在
database2.table3
Ddl
datarkRead
md_customer_tage_info
ChangeLog
STOPPED
报警
Config 流
本地表
匹配表名进行分流
APPEND/REPLACE/UPSERTDataSink
FAILED
Cluster1b
分布式表
父OutputFormat初始化连接数据源
创建子OutputFormat/OpenOutputFormat
父子OutputFormat
轮询拉配置
多对一OutputFormat
database1.table2
Database.TablenameDataSource
customer_id 去重流(windows:10swartermark:30s)
