idmapping(用户唯一标识)第二天数据生成

/**
  *思想逻辑:(必须整明白!!!)
  * 我们是要考虑第二天的数据进来了
  * 在拿去跟第一天的数据一起进行计算
  * 我们需要改的地方就是在构造图的地方添加昨天的点集合和边集合
  *
  * 1.将昨天的字典解析成点集合,边集合,
  * 2.将今天的点集合,边集合union到昨天的点边集合
  * 3.用union之后的点边集合构造最大连同子图
  * 4.调整结果
  *   今天有一套结果,昨天也有一套结果,
  *   虽然两套结果都对,但是总有可能今天的结果新生成的guid和昨天的有一点点小误差,
  *   为了方便日后留存查询,我们将今天有小误差的guid改成昨天的
  *   只有这样才能保证今天生成的guid和昨天生的guid以及以后所有的guid才能使同一个人
  *   这样我们以后查询当月的人流量才能确定是同一个人,避免泡沫数据和虚假数据!!!!
  *   步骤实现:
  *   4.1:我们将今天guid的行组合在一起,将昨天的字典搞成一个hashmap进行广播
  *   4.1:拿今天字典的点集合和昨天的额点集合求交集取到谁今天的guid就换成昨天的guid
  *
  */
object idmapping_nextday {
    
  def main(args: Array[String]): Unit = {
    
    //1.1.创建spark环境
    val spark: SparkSession = SparkSession.builder()
      .appName(this.getClass.getSimpleName)
      .master("local[*]")
      .getOrCreate()
    //将rdd变成df
    import spark.implicits._
    //2.导入数据
    val applog: Dataset[String] = spark.read.textFile("F:\\yiee_logs\\2020-01-11\\app\\doit.mall.access.log.8")

    //3.处理数据
    //构造一个点集合
    val data: RDD[Array[String]] = applog.rdd.map(line => {
    
      //将每行数据解析成json对象
      val jsonObj = JSON.parseObject(line)

      // 从json对象中取user对象
      val userObj = jsonObj.getJSONObject("user")
      val uid = userObj.getString("uid")

      // 从user对象中取phone对象
      val phoneObj = userObj.getJSONObject("phone")
      val imei = phoneObj.getString("imei")
      val mac = phoneObj.getString("mac")
      val imsi = phoneObj.getString("imsi")
      val androidId = phoneObj.getString("androidId")
      val deviceId = phoneObj.getString("deviceId")
      val uuid = phoneObj.getString("uuid")

      Array(uid, imei, mac, imsi, androidId, deviceId, uuid).filter(StringUtils.isNotBlank(_))
    })

    //构造一个点集合
    // 将每一个元素和他的hashcode组合成对偶元组
    val vertices: RDD[(Long, String)] = data.flatMap(arr => {
    
      for (biaoshi <- arr) yield (biaoshi.hashCode.toLong, biaoshi)
    })

    //构造图计算的边集合
    // Edge参数一:第一个点.参数二:下一个点.参数三:边的数据或者边的属性,这里我们用空串表示
    //将两个点连接起来
    val edges: RDD[Edge[String]] = data.flatMap(arr => {
    
      //用双重for循环的方法让数组中所有的两两组合成边
      for (i <- 0 to arr.length - 2; j <- i + 1 to arr.length - 1) yield Edge(arr(i).hashCode.toLong, arr(j).hashCode.toLong, "")
    })
      //然后在统计每一条边出现的次数
      .map(edge => (edge, 1)).reduceByKey(_ + _)
      //过滤将重复次数<5(经验阈值)的边去掉,
      .filter(tp => tp._2 > 2)
      .map(x => x._1)
    edges



    // 五、将上一日的idmp映射字典，解析成点、边集合
    val schema = new StructType()
      .add("biaoshi_hashcode",DataTypes.LongType)
      .add("guid",DataTypes.LongType)


    //新0将昨天的节过读出来
    val preDayIdmp: DataFrame = spark.read.parquet("data/dict/idmapping_dict/2020-02-15")
    //新1.读取昨天的字典解析成点集合
    val preDayIdmpVertices = preDayIdmp.rdd.map({
    
      case Row(idFlag: VertexId, guid: VertexId) =>
        (idFlag, "")
    })

    //新2.读取昨天的字典解析成边集合
    val preDayEdges = preDayIdmp.rdd.map(row => {
    
      val idFlag = row.getAs[VertexId]("biaoshi_hashcode")
      val guid = row.getAs[VertexId]("guid")
      Edge(idFlag, guid, "")
    })

    //新3.用 点集合 和 边集合 构造一张图  使用Graph算法和昨天的结果union在一起
    val graph = Graph(vertices.union(preDayIdmpVertices),edges.union(preDayEdges))

    //并调用最大连同子图算法VertexRDD[VertexId] ==>rdd 里面装的元组(Long值,组中最小值)
    val res_tuples: VertexRDD[VertexId] = graph.connectedComponents().vertices
    //最终我们要输出的文件是parquet的文件






    // 最后的调整结果、将结果跟上日的映射字典做对比，调整guid
    // 1.将上日的idmp映射结果字典收集到driver端，并广播
    val idMap = preDayIdmp.rdd.map(row => {
    
      val idFlag = row.getAs[VertexId]("biaoshi_hashcode")
      val guid = row.getAs[VertexId]("guid")
      (idFlag, guid)
    }).collectAsMap()
    val bc = spark.sparkContext.broadcast(idMap)

    // 2.将今日的图计算结果按照guid分组,然后去跟上日的映射字典进行对比
    val todayIdmpResult: RDD[(VertexId, VertexId)] = res_tuples.map(tp => (tp._2, tp._1))
      .groupByKey()
      .mapPartitions(iter=>{
    
        // 从广播变量中取出上日的idmp映射字典
        val idmpMap = bc.value
        iter.map(tp => {
    
          // 当日的guid计算结果
          var todayGuid = tp._1
          // 这一组中的所有id标识
          val ids = tp._2

          // 遍历这一组id，挨个去上日的idmp映射字典中查找
          var find = false
          for (elem <- ids if !find) {
    
            val maybeGuid: Option[Long] = idmpMap.get(elem)
            // 如果这个id在昨天的映射字典中找到了，那么就用昨天的guid替换掉今天这一组的guid
            if (maybeGuid.isDefined) {
    
              todayGuid = maybeGuid.get
              find = true
            }
          }

          (todayGuid,ids)
        })
      })
      .flatMap(tp=>{
    
        val ids = tp._2
        val guid = tp._1
        for (elem <- ids) yield (elem,guid)
      })
    // 可以直接用图计算所产生的结果中的组最小值，作为这一组的guid（当然，也可以自己另外生成一个UUID来作为GUID）
    import spark.implicits._
    // 保存结果
    todayIdmpResult.coalesce(1).toDF("biaoshi_hashcode", "guid").write.parquet("data/dict/idmapping_dict/2020-02-16")



    spark.stop()
  }
}