以flink实时流的方式实现OneId

2023-01-30 05:18 由 sqhhhhAA111i 发表于 #数据库

前言

oneid相关概念,及其当前离线实现方式介绍请见以下链接及其系列其他文章:

读完上述链接的相关文章,将理解oneid的需求及其实现方式.

之前公司是做电商数据分析,可以接入多个数据源的数据(美团,饿了么,京东等),在我们系统中,我们将为每个用户统一打上在我们系统内部的user_id,即oneid.当时已经有了基于spark图计算实现的id-mapping来达成oneid.

到这里,我们已有的系统跟前言中链接文章提到的内容相似.

后来,我们有了新需求:实时化改造.实时为每个用户统计相关数据.

但是,实时计算的数据最后要归纳到用户上,那么我们的id-mapping也需要实时出现结果.

本人不具备机器学习领域的知识,在看了几遍前言中的文章后,勉强搞懂了该图计算的原理,本质上是求最大连通图,研究后发现需要拿到全量数据进行迭代,不能改造为实时计算.

后来,基于"最大连通图"的算法,变种出了一种方法:

输入两个数据关联关系,例如美团id1-饿了么id1,即输入两个点一个线.
拿到该关系,分别拿两个数据去与已有的id-mapping结果表对比.如,我们结果id-mapping表中,有关系
```
美团id1 - oneid1
饿了么id1 - oneid2
```
通过第一步中传入的关系,可以得出美团id1 与饿了么id1 在我们系统中应该识别为同一个人,对应同一个oneid,可以得到
```
美团id1 - oneid1
饿了么id1 - oneid1
```
或者,都对应为另一个oneid
```
美团id1 - oneid2
饿了么id1 - oneid2
```
即,我们通过传入关联关系,将美团id1 与饿了么id1 在我们系统中重新更新为关联到同一个oneid.
后续,某个用户id是美团id1,那么它关联这个id-mapping结果表,可以得到它在我们系统中的id是oneid1(或者oneid2,此处根据第三步如何取值)
根据不同的对比结果,进行相应的替换或者新增,我们变相实现了"最大连通图"的算法,并且这个算法可以用flink实时计算实现

结果表可以有多个描述字段,但是核心应该是以下两个字段:

原id , 计算出的oneid

我们在采集数据的时候,需要将数据解析成两两的关系对.如原始数据:

手机号1,美团id1,设备id1

需要将这条消息拆分为:

手机号1 - 美团id1
手机号1 - 设备id1
美团id1 - 设备id1

再将这三组关联关系传给后续对比计算.

假设我们得到关系对:

x - y

我们拿到此关系对到结果表中进行对比将有以下几种情况:

x,y都没有对应oneid: 直接对结果表插入计算得出的新oneid(可以使用uuid)
```
x - 新oneid
y - 新oneid
```
x已有对应oneid为 XXoneid,y没有:将y的oneid赋值为 XXoneid,并插入,得到
```
x - XXoneid
y - XXoneid
```
x没有,y有oneid为 YYoneid :同第二种情况,得到
```
x - YYoneid
y - YYoneid
```
x,y都有oneid,且一致,都为 ZZoneid: 不更新

x,y都有oneid,且不一致,分别为 XXoneid,YYoneid :将 x,y更新为同一个oneid(XXoneid或者YYoneid),或者重新生成一个.此处看个人选择.

并且!!!!!!
将结果表中所有oneid为 XXoneid,YYoneid的相关数据,oneid都重设为新选择的oneid
这是为了将相关联的其他数据一起指向新的oneid

至此,通过以上几种情况.我们复现了id-mapping中求最大连通图的算法.

kafka

flink

redis:将结果表以k-v的形式放在内存中,这样flink可以快速取值并对比计算

hbase:此处可以换为mysql,doris等支持更新的存储即可.并且还有以下原因:

对比计算中,第五种情况,需要从这里取所有oneid为 XXoneid,YYoneid的相关数据
而redis中没法根据value来取得key,所以第五种情况,需要查询此处存储得到相关数据

结果不但要更新hbase,还要更新redis中存放的k-v对!!!建议先更新redis,因为比较快.