计算one id集成主数据流程图

如果没有，只插入customer_base_info表中one_id和身份证信息（如果type为身份证）

标记before

根据最新生成的one id删除主数据表中数据，防止数据重复customer_base_phone、customer_base_email、customer_base_info(由于产品信息source数据源和result数据源公用一张表，所以这张表只需要赋值one id即可，前面已经赋值过)

【集成邮箱】转换生成客户邮箱主数据，邮箱加密，赋值最新one id，插入customer_base_email表中

过滤未删除&未解绑&非脏数据

去重

【集成手机信息】根据org_id、busi_id、cus_id从ods_customer_contact数据源表查询客户联系信息，转换生成客户手机主数据，赋值最新one id，插入customer_base_phone主数据表中

过滤空值

flatMap

否

集成主数据-先删后增

考虑并发计算的可能，如果customer_id表中one_id\\type\\value\\busi_id\\org_id\\cus_id相同，那么说明计算one id后的值是重复的，需要删除重复的数据

取org_id和cus_id集合中最小值，拼接成最新的one id，这样处理的优势是保证org_id和cus_id永远处于最父级

输出one id

如果有且仅有一条数据，取当前记录

【集成基础信息】转换生成客户基础信息主数据，赋值最新one id，插入customer_base_info表中

获取历史one id集合，去重

将所有one id拆分成org_id 和 cus_id集合

加入after数据对应的org_id和cus_id

source源ods层表：    ods_customer    ods_customer_contact    ods_customer_source(one id计算采集数据源)one id结果id mapping映射表：    customer_id_mapping主数据表：    customer_base_info    customer_base_phone    customer_base_email    customer_product_info

阈值计算，匹配source数据源，查询ods_customer、ods_customer_contact表

遍历

重复

1个手机号超过10人使用？

如果为after数据，直接将数据流中id mapping数据插入customer_id_mapping表中

转换IdMapping对象流Id Mapping Data Streaming

after数据：新增动作，计算这条数据的one id

【转换】将source(json)数据转换成id mapping(entity)数据

如果为before数据，根据org_id、busi_id、cus_id、type、value删除customer_id_mapping表中数据

根据历史one id，更新主数据表customer_base_email、customer_base_info、customer_base_phone、customer_product_info表中数据为最新one id

结束遍历

End

身份证超过18位？

超过

org_id、busi_id、cus_id集合

Start

create：拥有after数据

before数据：删除动作，剔除这条数据的one id

map

read：拥有after数据

【集成产品信息】由于和ods数据源公用同一张表数据，这里只需要根据org_id、busi_id、cus_id更新customer_product_info表中one id字段数据即可

1个身份证超过5人使用？

标记after

根据org_id、busi_id、cus_id从ods_customer数据源表查询客户基础信息，这里应该只会有一条数据

忽略计算one id，标记source数据源为脏数据

如果有多条数据，取最新事件时间的一条数据

根据历史one id，更新customer_id_mapping表中数据为最新one id

delete：拥有before数据

flink-cdc读取ods_customer_source表数据（实时或离线）

update：拥有before、after数据，不做任何操作，否则标记脏数据会造成死循环和逻辑冲突

1个用户只有一个身份证？