Spark DataFrame API 踩坑记

1. 当join 条件使用 df("col_1") === df2("col_2")， 由于spark sql 执行计划的原因，有可能会导致结果出现一异常

解决方案： 使用sql 或者 重命名为一样的字段名字

 @Test
  def joinTest() = {
    spark.conf.set("spark.sql.crossJoin.enabled", true)
    spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
    spark.conf.set("spark.sql.orc.filterPushdown", false)

    val source = Seq(
      ("11111111",  Seq("22222222", "33333333")),
      ("22222222",  Seq("44444444", "33333333")),
      ("33333333",  Seq("44444444", "22222222")),
      ("44444444",  Seq( "22222222"))
    )

    val rawDF = spark.createDataframe(source).toDF("esn", "esn_list")

    rawDF.createOrReplaceTempView("raw_tb")

    rawDF.printSchema()
    rawDF.show(false)

    val tmpDF = rawDF.select(
      col("esn")
      , explode(col("esn_list")).as("group_esn")
    )
    tmpDF.createOrReplaceTempView("tmp_db")

    tmpDF.printSchema()
    tmpDF.show(false)

    val rawTmpDF = rawDF.selectExpr("esn as group_esn", "esn_list")
    //val rawTmpDF = rawDF.selectExpr("esn as esn2", "esn_list")

    // val testDF = tmpDF.join(rawDF, rawDF("esn") === tmpDF("group_esn"), "inner") // 结果为空



    val testDF = tmpDF.join(rawTmpDF, rawTmpDF("group_esn") === tmpDF("group_esn"), "inner")
// val testDF = tmpDF.join(rawTmpDF, rawTmpDF("esn2") === tmpDF("group_esn"), "inner")

    testDF.printSchema()
    testDF.show(false)

    testDF.explain(true)

// sql 正常没有问题
//    val sql =
//      """
//        |select *
//        |from raw_tb t1
//        |inner join tmp_db t2
//        |on t1.esn = t2.group_esn
//        |
//        |""".stripMargin
//
//    val frame = spark.sql(sql)
//
//    frame.printSchema()
//    frame.show(false)
//    frame.explain(true)


  }