太一项目图计算SparkGraphX思维导图

样例类根据数据的头文件进行生成

case class UserEventLog(username: String,
                          phone: String,
                          mac: String,
                          imei: String,
                          open_id: String,
                          android_id: String,
                          idfv: String)

df_20240706

Spark.read

val user_event_log_20240706: Dataset[UserEventLog] = df_20240706.as[UserEventLog]

 按照分区在处理每一条

过滤null

缓存

val idsCache_20240706 = user_event_log_20240706.rdd
      //按照分区处理再处理每一条
      .mapPartitions(partIter => {
        partIter.map(event => {
          Array(event.username, event.phone, event.mac, event.imei, event.open_id, event.android_id, event.idfv)
            //过滤null
            .filter(StringUtils.isNotBlank)
        })
        //缓存
      }).persist(StorageLevel.MEMORY_AND_DISK)

把顶点属性进行哈希 得到顶点值

循环遍历所有顶点

yield 集合生成器

去重

// 对每个 Array 进行处理并创建顶点集合
    // 构建点集合 哈希 zhangsan 获取 顶点 顶点的属性就是zhangsan
    val vertices_20240706: RDD[(VertexId, String)] = idsCache_20240706
      .mapPartitions(partIter => {
        // 建议采用碰撞率更低效率更高的 Hashing.murmur3_128()
        val hashFunction = Hashing.murmur3_128()
        partIter.flatMap(arr => {
          // (点的唯一标识, 点的数据)  传进来的值是String 所以用hashString  i是属性zhangsan  (hash,zhangsan)
//          yield 集合生成器
          for (i <- arr) yield ((hashFunction.hashString(i, StandardCharsets.UTF_8).asLong.abs), i)
        })
      })
//      去重
      .distinct()

创建临时视图

 源顶点

目标顶点

边数据

去重

// 对每个 Array 进行处理并创建边集合
    val edges_20240706: RDD[Edge[String]] = idsCache_20240706
      .mapPartitions(partIter => {
        // 建议采用碰撞率更低效率更高的 Hashing.murmur3_128()
        val hashFunction = Hashing.murmur3_128()
        partIter.flatMap(arr => {
          // Edge(srcId：源顶点，dstId：目的顶点，attr：边数据)
          for (i <- 0 to arr.length - 2; j <- i + 1 to arr.length - 1)
            yield Edge(
              hashFunction.hashString(arr(i), StandardCharsets.UTF_8).asLong.abs,
              hashFunction.hashString(arr(j), StandardCharsets.UTF_8).asLong.abs,
              ""
            )
        })
      })
//      重复去重
      .distinct()

 顶点集合对象和边集合对象

val connectedSubgraph_20240706 = graph_20240706.connectedComponents()

val newIdMapping = spark.sql(
      """
        |SELECT t1.id_hashcode AS id_hashcode, t2.id AS id, t1.guid AS guid
        |FROM subgraph t1
        |LEFT JOIN vertex_20240706 t2 ON t1.id_hashcode = t2.id_hashcode
        |GROUP BY t1.id_hashcode, t2.id, t1.guid;
        |""".stripMargin)
    // 将结果按日期分区，写入 GUID 维度字典表 dim.dim_guid_dict

//之前用的 是子图的最小hashcode  从新生成一个guid
    // GUID 也可以采用 MD5/UUID 的方式
    spark.sql(
      """
        |SELECT t1.id_hashcode AS id_hashcode, t2.id AS id, MD5(CAST(t1.guid AS STRING)) AS guid
        |FROM subgraph t1
        |LEFT JOIN vertex_20240706 t2 ON t1.id_hashcode = t2.id_hashcode
        |GROUP BY t1.id_hashcode, t2.id, t1.guid;
        |""".stripMargin)
    //.show(100, truncate = false)

val dim_guid_dict: DataFrame = spark.read
      .option("sep", ",")
      .option("inferSchema", "true")
      .csv("file:///E:\\Projects\\IdeaProjects\\yjxxt\\tydt-offline\\tydt-offline-etl\\src\\test\\resources\\data\\day01_idmapping")
//      三列
      .toDF("id_hashcode", "id", "guid")

val vertices_20240706: RDD[(VertexId, String)] = dim_guid_dict.rdd
      .mapPartitions(partIter => {
        partIter.map(row => (row.getAs[VertexId]("id_hashcode"), row.getAs[String]("id")))
      })

还是分区按行读取

上一日的维表的id_hashcode，id 组合起来就是顶点集合

第一步：提取（guid，id_hashcode）键值对

第二步：按 guid 分组，聚合同一用户的所有 id_hashcode

第三步：为同一 guid 下的所有 id_hashcode 生成两两相连的边

 按第一日guid分组→同一组内 ID 两两连边” 的逻辑，复现了第一日的用户关联关系，为第二日的图计算提供了历史关联基础，最终保证跨日的用户唯一标识（guid）的一致性。

val allEdges = edges_20240707.union(edges_20240706)

val graph = Graph(allVertices, allEdges)

val connectedSubgraph = graph.connectedComponents()

 提取关联之后的数据

val connectedSubgraphVertices_20240707 = connectedSubgraph.vertices.toDF("id_hashcode", "guid")

将上一日 dim_guid_dict 的数据，收集到 Driver 端，然后作为变量广播出去

如果广播变量非常大，广播的时候可能会存在 OOM 的风险

从第一日的 GUID 维度字典表dim_guid_dict中提取(id_hashcode, guid)映射关系，转换为 Map 集合

通过广播变量bcIdMappingMap将该映射关系分发至所有 Executor，避免分布式计算中频繁传输大表数据，提升查询效率。

// 遍历 ID，挨个映射查找  把第二天的进行guid分组 然后拿着数组里面的数据去第一天找guid  用第一天的guid 去覆盖第二天的guid

分组：将第二日连通子图结果按临时guid分组，得到(tempGuid, Iterable[id_hashcode])，即每个临时子图包含的所有 ID 哈希值。

修正逻辑：对每个临时子图，遍历其包含的id_hashcode，若存在于第一日的映射关系中，则将该子图的guid更新为第一日对应的guid（确保跨日用户标识一致性）；若不存在，则保留临时guid（新增用户）。

扁平化：将(finalGuid, Iterable[id_hashcode])转换为(id_hashcode, finalGuid)的扁平结构，得到修正后的id_hashcode与guid映射。

// 将这一日结果跟上一日的 GUID 维度字典表数据做对比，更新 GUID
    val newIdMapping = connectedSubgraphVertices_20240707.rdd
      .map(row => (row.getAs[VertexId]("guid"), (row.getAs[VertexId]("id_hashcode"))))
      .groupByKey() // 按 guid 分组
      .mapPartitions(partIter => {
        partIter.map(kv => {
          // 获取广播变量中上一日 dim_guid_dict 的数据  优化点
          val idMappingMapForExecutor = bcIdMappingMap.value
          var newGuid = kv._1
          val ids_hashcode = kv._2
          // 遍历 ID，挨个映射查找  把第二天的进行guid分组 然后拿着数组里面的数据去第一天找guid  用第一天的guid 去覆盖第二天的guid
          breakable {
            for (id_hashcode <- ids_hashcode if idMappingMapForExecutor.contains(id_hashcode)) {
              newGuid = idMappingMapForExecutor(id_hashcode)
              break()
            }
          }
          // 还是返回 groupByKey 后的数据格式，只是 guid 已被更新
          (newGuid, ids_hashcode)
        })
      })

更新完的第二日维表之后  使用行为数据的ID标识去查这一日的GUID 维度字典表 找出对应的guid 回填到行为数据

再将第二日的行为数据中的guid 和第一次的GUID维表字典表使用guid 进行LEET JOIN JOIN上的就是老用户 JOIN不上的就是新用户