社交平台分类

任务1：将数据库中的粉丝历史关注数据follow_00 ~ follow_09批量导入到Neo4j中，这个任务在初始化的时候只需要执行一次。
任务2：通过Spark实时维护粉丝的关注与取消关注数据，它的数据源自于Kafka中的user_follow这个Topic，它的计算结果也保存到Neo4j。
任务3：通过Spark每天计算并更新用户的活跃时间，如果当天活跃过，就更新最近活跃时间为当前日期，这是针对主播和粉丝都会进行的计算任务，它的数据源自于HDFS中的user_active，计算结果保存到Neo4j。
任务4：通过Spark每天计算并更新主播等级数据，这项计算任务仅针对主播，它的数据源自于HDFS中的user_level，计算结果保存到Neo4j。
任务5：通过Spark每周计算主播最近30天的内容评级，它的数据源自于HDFS中的live_info，计算结果保存到Neo4j。
任务6：通过Spark每周计算主播近一个周的三度关系列表，参与计算的主播需要满足下面几个前提条件。
- 主播近一周内处于活跃状态。
- 主播等级 > 10。
- 主播近30天至少3条内容评价满足A+。
- 主播粉丝列表关注重合度 > 2。
任务7：每周将计算结果导出到MySQL。

添码座原创大约 2 分钟

任务1：导入历史关注数据

将从MySQL中导出的历史关注数据导入到Neo4j。

# 将之前导出的数据文件follower_0${i}.log拷贝到neo4j的import目录
> cp /var/logs/follower_*.log /home/work/neo4j-community-5.23.0/import

> cd /home/work/neo4j-community-5.23.0
> ./bin/cypher-shell -a neo4j://172.16.185.176:7687 -u root -p 12345678

# 创建索引
neo4j> CREATE CONSTRAINT ON (u:User) ASSERT u.uid IS UNIQUE;

# 逐个导入数据
neo4j> LOAD CSV FROM 'file:///follower_00.log' AS row
       MERGE (viewer:User {uid: toString(row[1])})
       MERGE (anchor:User {uid: toString(row[2])})
       MERGE (viewer) -[:follow]-> (anchor);

neo4j> LOAD CSV FROM 'file:///follower_01.log' AS row
       MERGE (viewer:User {uid: toString(row[1])})
       MERGE (anchor:User {uid: toString(row[2])})
       MERGE (viewer) -[:follow]-> (anchor);

......

neo4j> LOAD CSV FROM 'file:///follower_09.log' AS row
       MERGE (viewer:User {uid: toString(row[1])})
       MERGE (anchor:User {uid: toString(row[2])})
       MERGE (viewer) -[:follow]-> (anchor);

添码座原创小于 1 分钟

任务2：维护实时关注数据

这一部分的数据来源于Kafka中的user_follow这个Topic，它的计算结果会被保存到Neo4j。

它的数据格式如下。

{
    "fid":"1",
    "uid":"2",
    "time":1718102763768,
    "type":"follow", # follow表示关注，或者为unfollow，表示取消关注
    "desc":"follow"  # 同上
}

添码座原创大约 6 分钟

数据采集配置

创建Topic

部署并启动ZooKeeper和Kafka后，就可以在Kafka中创建需要的Topic了。

添码座原创大约 4 分钟

采集数据库数据

历史关注数据使用shell脚本导出MySQL中的数据，然后再存储到HDFS。

而博主或主播的等级数据是需要每天导一次的，也可以用shell脚本导入。

添码座原创大约 1 分钟