关联图谱特征计算

讨论完流数据在时间维度的聚合分析，我们再来看看流数据在空间维度的聚合分析，也就是关联图谱。

关联图谱是一种使用“图”来表示实体之间关联关系的数据组织结构。在社交网络进行分析中，关联图谱有着广泛的应用。通过对社交网络进行分析，可以发现虚拟社区、评估个体影响力、探索信息传播规律等。

图4-8展示了一个关联图谱的例子，将这个关联图谱可视化后，我们能够一目了然地发现该图谱中有3个“团伙”，每个“团伙”各有1到2个“大哥”，并且3个“团伙”之间还通过“小弟”相互联系。

图4-8 关联图谱

同样以金融风控为例，关联图谱在其中扮演着重要角色并起着巨大的作用。例如，在游戏代充值场景中，通过对手机和用户构成的网络分析，发现某个手机上注册的不同用户数过多，说明这个手机非常可疑。再如，在反欺诈场景中，通过对IP和设备的网络分析，发现某个IP C段上出现的设备数过多，说明这个IP C段的网络可能是团伙欺诈网络。

在本节中，我们主要讨论关联图谱中一度关联和二度关联的特征计算问题。虽然本节主要讲解流数据在空间维度的聚合分析，但是由于流数据本身属于时间序列，并且具有无穷无尽的特点，我们还是需要将时间窗口考虑在内。具体来说，我们分析的问题是诸如“过去一周内在同一个设备上注册的不同用户数”“过去24小时同一IP C段220.181.111出现的不同设备数”这类有时间窗口限定的问题，而不是“同一个设备上注册的不同用户数”和“同一IP C段220.181.111出现的不同设备数”这种不设时间范围的问题。

4.4.1 一度关联

一度关联是指关联图谱中的一个节点有多少个与之直接相邻的节点。实时流上的一度关联通常是为了统计一段时间内，某种属性上另一种属性不同取值的个数。例如，“过去一周内在同一个设备上注册的不同用户数”“过去24小时同一IP C段220.181.111出现的不同设备数”“过去1小时用户账户使用的不同IP数”“过去3个月同一手机号码关联的不同设备数”等。

同样，如果用SQL来描述这类问题，就应该是类似于以下这些例子：

# 过去一周内在同一个设备上注册的不同用户数

SELECT COUNT(DISTINCT user_id) FROM stream

WHERE event_type = "create_account"

AND timestamp >= 1530547200000 and timestamp < 1531152000000

GROUP BY device_id;

# 过去1小时用户账户使用的不同IP数

SELECT COUNT(DISTINCT ip) FROM stream

WHERE event_type = "transaction"

AND timestamp >= 1531065600000 and timestamp < 1531069200000

GROUP BY user_id;

# 过去3个月同一手机号码关联的不同设备数

SELECT COUNT(DISTINCT device_id) FROM stream

WHERE event_type = "create_account"

AND timestamp >= 1530547200000 and timestamp < 1538496000000

GROUP BY phone_number;

从上面的示例中可以看到，一度关联的计算其实就是COUNTDISTINCT（去重计数）计算。所以，我们立刻就想到了在流数据中实现一度关联的方法。

首先，我们在每个时间窗口内，用一个集合（set）来记录变量所有不同的取值。当新事件到达时，将事件所带相关变量的值添加到集合，利用集合自身的特性实现去重功能，然后返回集合的势（也就是集合的大小），即我们要计算的一度关联特征值。是不是非常简单啊？当读者看到此处时，就知道按照笔者的惯例，这其中必定存在猫腻了。是的，这里也存在问题，而且问题还不小。

针对一度关联的计算，我们的目的是得到一度相邻节点的数量。在前面的解决方案中，为了实现这个目标，我们非常朴实地将每一个不同取值都放在集合中保存下来。在数据量较小的时候，这种做法简单明了，不会存在什么问题。但如果变量的势很大，不同取值非常多，那么保存这些值将会占用大量的存储空间。不仅如此，当数据量大到一定程度时，程序的实时计算性能急剧下降。占用大量的存储空间和衰减严重的性能表现，都会让前面的解决方案在实际生产环境中变得不可行。

那该怎么办呢？在这关键时刻，一位盖世英雄踩着七彩祥云来拯救我们了，它就是神奇的HyperLogLog算法！话说笔者当初为了解决一度关联特征计算的问题可谓费尽心机，后来在一次上网查阅时搜得HyperLogLog算法，顿觉惊为天人，继而拍案而起，这种算法真的是太神奇了！

HyperLogLog算法是一种以准确度换取时间复杂度和空间复杂度的近似算法，与之类似的还有Bloo.Filter、Count-Mi.Sketch等算法。我们接下来重点介绍的HyperLogLog算法就是为了解决大数据量情况下计算集合中去重元素个数的问题。HyperLogLog能够帮助我们节省大量存储空间和计算时间。以Redis中的HyperLogLog算法实现为例，只需要用12KB的内存，就能够在0.81%的标准误差范围内，记录将近264个不同值的个数。而如果我们将这些不同值原原本本地记录下来，那就是平均记录长度×264B了。另外，HyperLogLog算法的插入和查询的时间复杂度都是O（1），所以在时间性能方面，HyperLogLog算法完全符合实时计算的要求。

在Redis中，HyperLogLog算法提供了3个命令：PFADD、PFCOUNT和PFMERGE。其中，PFADD用于将元素添加到HyperLogLog寄存器，PFCOUNT用于返回添加到HyperLogLog寄存器中不同元素的个数（根据HyperLogLog算法计算出来的估计值），PFMERGE则用于合并多个HyperLogLog寄存器。

在有了HyperLogLog算法的加持后，我们就能够对一度关联的计算做出优化了。首先，我们在每个时间窗口内，为变量创建一个新的HyperLogLog寄存器。当新事件到达时，将事件所带相关变量的值通过PFADD命令添加到HyperLogLog寄存器中，然后使用PFCOUNT命令就可以返回变量不同取值的数量（估计值），这就是一度关联值。而如果我们还需要对多个窗口内的不同值个数进行汇总，那么就使用PFMERGE命令先将多个窗口内的HyperLogLog寄存器合并起来，生成一个新的合并后的HyperLogLog寄存器，之后对这个寄存器使用PFCOUNT命令就可以返回合并多个窗口后变量的不同取值个数了。

当然，虽然HyperLogLog算法为我们解决了去重计数的问题，但是还存在与4.3节进行时间维度聚合计算时一样的问题，即如果计算一度关联的分组变量（如本小节前文所述3个SQL示例中的device_id、user_id和phone_number）本身就有非常高的势，那么就需要非常非常多的HyperLogLog寄存器。如果按照每个HyperLogLog寄存器12KB计算，其实这也是一笔不小的存储空间开销了。所以，同样的道理，我们需要将这些寄存器放到诸如Redis、Ignite或本地磁盘这样的外部存储器中，并且为这些寄存器设置过期时间。另外，如果你能够接受更大的估计误差，则还可以进一步减小HyperLogLog寄存器的长度。

表4-2列举了使用不同长度HyperLogLog寄存器情况下1000万个寄存器占用的空间及对应的估计误差。

表4-2　不同长度HyperLogLog寄存器占用的空间与估计误差

 

注：估计误差ERR与寄存器长度L（以B为单位）之间的关系为ERR=1.04/sqrt（L*8/6），其中，8为1B对应的位数，6为HyperLogLog算法中每个桶使用的位数。

4.4.2 二度关联

二度关联是对一度关联的扩展，它是由节点的一度关联节点再做一次一度关联后的节点数，如“过去一个月内在同一个设备上注册的用户登录过的设备数”“过去一个周内来自于同一个IP的设备使用过的IP数”。图4-9描述了一个节点的二度关联节点，其中所有标记为1的节点都是标记为0的节点的一度关联节点，而所有标记为2的节点都是标记为0的节点的二度关联节点。

从图4-9中，我们能够直观地理解到，要计算一个节点的二度关联节点数，需要执行两个步骤。第一步是获取该节点的所有一度关联节点所组成的集合。第二步是遍历这个集合，获取其中每个节点的一度关联节点所组成的集合，然后将所有这些集合求并集。最后得到的这个并集就是原节点的二度关联节点集合了。由于二度关联这种天生的“两步走”过程，我们在实现二度关联的计算时，也将这两个步骤分开。第一步是求一个集合，第二步则与4.4.1节中一度关联的计算类似。

图4-9 二度关联

讨论到这里的时候，聪明的读者们也一定发现了二度关联计算的问题。我们在4.4.1节中就谈到过，为了避免过多占用存储空间及性能随时间的衰减，我们采用了不需要记录原始值的HyperLogLog算法。可是当涉及二度关联计算的时候，我们不可避免地需要记录位于原节点和二度关联节点之间的一度关联节点。毫无疑问，如果一度关联节点很多，则这个方案就不可行了。

实际上，我们这次是真的遇到挑战了。在实时流计算领域，目前尚且没有一种在大数据量情况下方便、直接且行之有效的二度关联计算方案。虽然有很多图数据库（如JanusGraph和Dgraph）在分布式实时图计算方面已经有了非常大的突破，能够帮助我们在一定程度上解决二度关联实时计算的问题，但相比实时流计算对响应时延及吞吐力更严苛的要求，还是略显不足。

所以我们完全没辙了吗？这也未必。如果我们愿意接受一个稍有滞后的二度关联计算结果，则我们还是能够采取一定的手段，做到二度关联的实时查询的。那究竟是什么方法呢？咱们就不卖关子了，它就是大名鼎鼎的Lambda架构！在第9章中，我们还会讨论Lambda架构，但在此我们先就二度关联这个具体的问题来先看看Lambda架构是如何发挥作用的。

Lambda架构的核心思想是对于计算量过大或者计算过于复杂的问题，将其分为离线计算部分和实时计算部分，其中离线计算是在主数据集上的全量计算，而实时计算则是对增量数据的计算。当这两者各自计算出结果后，再将结果合并起来，从而得到最终的查询结果。通过这种离线计算和实时计算的方式，Lambda架构能够实时地在全量数据集上进行分析和查询。

对于二度关联计算，我们也将其分为离线计算部分和实时计算部分。下面就以“过去一个月内在同一个设备上注册的用户登录过的设备数”这个计算目标为例，详细讲解具体实现方法。

首先，将流数据按照不同的事件类型，存入不同的Hive表中。在“过去一个月内在同一个设备上注册的用户登录过的设备数”这个特征计算中，我们将注册（create_account）事件存放到create_account_table表中，将登录（login）事件存放到login_table表中。这两个表的定义分别如下：

CREATE TABLE create_account_table(device_id string, user_id string) PARTITioNED BY

(day string, hour string);

CREATE TABLE login_table(user_id string, device_id string) PARTITIONED BY (day

string, hour string);

接下来，我们先计算离线部分的不同设备数。假设每次大约需要120分钟才能执行完一个月的数据，再留下一部分空档时间，于是我们设定每3小时执行一次离线计算。例如，在2019/09/3.09：03：00时刻，开始执行如下离线计算部分的Hiv.SQL。

-- 每3小时执行一次

CREATE TABLE temp_table_before_20190930_09 AS

SELECT DISTINCT

create_account_table.device_id AS c_device_id,

create_account_table.user_id AS user_id,

login_table.device_id AS l_device_id

FROM

create_account_table INNER JOIN login_table ON create_account_table.user_id =

login_table.user_id

WHERE

(

create_account_table.day < "20190930" AND create_account_table.day >=

"20190901"

AND

login_table.day < "20190930" AND login_table.day >= "20190901"

)

OR

(

create_account_table.day = "20190930" AND create_account_table.hour < "09"

AND

login_table.day = "20190930" AND login_table.hour < "09"

);

在上面的Hiv.SQL中，我们将create_account_table表和login_table表通过共同的用户user_id关联起来，并通过DISTINCT关键字得到去重后的用户注册和登录设备信息，这样就得到了离线部分的计算结果。

接下来就是实时计算部分了。在实现实时计算部分前，我们需要先确定实时计算部分需要计算的内容，以及之后怎样将实时计算部分的结果合并到离线计算部分上来。图4-10展示了二度关联特征的增量计算方法，其中每个有向连线都代表了一部分数据之间的内联接（inne.join）操作。具体来说，A→B代表离线计算部分，剩下的ΔA→ΔB、ΔA→ΔB、ΔA→ΔB代表增量计算的部分。

图4-10 二度关联特征的增量计算方法

前面我们已经假定计算一个月的数据需要120分钟，每隔3小时计算一次。所以，实时计算部分最多需要计算最近6小时内的增量数据，再考虑每天不同时刻的流量是有高峰和低谷之别的，所以我们保守估计在实时计算部分，ΔA→B、A→ΔB分别需要4分钟，而ΔA→ΔB需要1分钟。算到这里，就有些尴尬了，实时计算部分居然需要9分钟，这还算实时计算吗？所以，我们在这里实现的实时计算部分是打了“大折扣”的，但不管怎样，将原本全量计算的时间从2个小时缩减为9分钟左右，也算是不小的进步了。

接下来就是实时计算部分的实现了，具体如下：

-- 计算ΔA→ΔB部分

CREATE TABLE temp_table_after_20190930_09_p1 AS

SELECT DISTINCT

create_account_table.device_id AS c_device_id,

create_account_table.user_id AS user_id,

login_table.device_id AS l_device_id

FROM

create_account_table INNER JOIN login_table ON create_account_table.user_id =

login_table.user_id

WHERE

(

create_account_table.day = "20190930" AND create_account_table.hour >= "09"

AND

login_table.day = "20190930" AND login_table.hour >= "09"

);

-- 计算A→ΔB部分

CREATE TABLE temp_table_after_20190930_09_p2 AS

SELECT DISTINCT

create_account_table.device_id AS c_device_id,

create_account_table.user_id AS user_id,

login_table.device_id AS l_device_id

FROM

create_account_table INNER JOIN login_table ON create_account_table.user_id =

login_table.user_id

WHERE

(

create_account_table.day < "20190930" AND create_account_table.day >=

"20190901"

AND

login_table.day = "20190930" AND login_table.hour >= "09"

);

-- 计算ΔA→B部分

CREATE TABLE temp_table_after_20190930_09_p3 AS

SELECT DISTINCT

create_account_table.device_id AS c_device_id,

create_account_table.user_id AS user_id,

login_table.device_id AS l_device_id

FROM

create_account_table INNER JOIN login_table ON create_account_table.user_id =

login_table.user_id

WHERE

(

create_account_table.day = "20190930" AND create_account_table.hour >= "09"

AND

login_table.day < "20190930" AND login_table.day >= "20190901"

);

在上面的SQL中，我们分别计算了ΔA→ΔB、ΔA→B、A→ΔB的增量数据。根据前面的分析，这部分执行需要9分钟左右，所以我们设定每15分钟执行一次以上SQL。

最后将离线部分和实时部分两者的结果合并起来：

SELECT c_device_id, COUNT(DISTINCT l_device_id)

FROM

temp_table_before_20190930_09

UNION temp_table_after_20190930_09_p1

UNION temp_table_after_20190930_09_p2

UNION temp_table_after_20190930_09_p3

GROUP BY c_device_id;

至此，我们就完成了“过去一个月内在同一个设备上注册的用户登录过的设备数”的统计。接下来可以将计算结果导入Redis缓存起来，以供流计算应用实时查询。

总的来说，在这种解决方案下，我们所查得的“过去一个月内在同一个设备上注册的用户登录过的设备数”是最多迟滞30分钟（由15分钟乘以2倍所得）的数据，但查询本身是实时快速响应的，毕竟只需要通过GET命令访问一次Redis即可。所以，不管怎样，这是一个可以真实落地且行之有效的解决方案。

最后，真心希望诸如JanusGraph和Dgraph等各种开源分布式图数据库[1]变得更加强大和丰富起来。毕竟，关联图谱分析本应该属于图数据库分内之事啊！感兴趣的读者不妨尝试下基于这些分布式图数据库的关联图谱分析方案，说不定就有意外惊喜呢！

[1] 注意，图数据库厂商TigerGraph 专门针对目前几种主流图数据库做过性能对比测试，感兴趣的读者可以自行查阅， 链接地址为https://www.tigergraph.com.cn/wp-content/uploads/2019/02/TigerGraph-Benchmark-Report-20190217.pdf。读者们可重点关注其中“两度路径查询时间”一表。

本篇文章给大家讲解的内容是数据处理： 关联图谱特征计算

下篇文章给大家讲解的内容是数据处理： 事件序列分析