Spark DataFrame 寫入 Phoenix 指定 Schema 避免自動生成 _1 _2 列名

問題現場

RDD 轉 DataFrame 寫入 Phoenix 表，列名自動變為 `_1`、`_2`... 導致與 Phoenix 表預定義列族 (HANGJIAN, LIECU...) 不匹配。

// 錯誤示範：RDD[Row] 無 Schema 推斷為 _1 _2...
val df = sqlContext.createDataFrame(rdd) // 自動生成 _1 _2 _3...
df.write.format("org.apache.phoenix.spark")...

根本原因

Spark 對無 Schema 的 `RDD[Row]` 或 `RDD[Tuple]` 推斷列名為 `_1`、`_2`... (Scala Tuple 命名慣例)。**必須顯式提供 Schema**。

解決方案：createDataFrame 重載方法指定列名

// 方案一：StructType Schema (最標準)
import org.apache.spark.sql.types._
val schema = StructType(Seq(
  StructField("HANGJIAN", StringType, nullable = false),
  StructField("LIECU", StringType, nullable = true),
  StructField("LIECU2", StringType, nullable = true),
  StructField("LIECU5", StringType, nullable = true),
  StructField("HANGJIAN5", StringType, nullable = true)
))
val df = sqlContext.createDataFrame(rdd, schema)

// 方案二：toDF 列名數組 (簡潔，需 RDD[Row] 順序一致)
val df = rdd.map(r => Row(r.getString(0), r.getString(1), ...))
           .toDF("HANGJIAN", "LIECU", "LIECU2", "LIECU5", "HANGJIAN5")

// 方案三：Case Class 編碼 Schema (Scala 最優雅)
case class PhoenixRecord(HANGJIAN: String, LIECU: String, LIECU2: String, LIECU5: String, HANGJIAN5: String)
val df = rdd.map(r => PhoenixRecord(r._1, r._2, r._3, r._4, r._5)).toDF()
// 隱式轉換自動推斷 Schema，列名即 Case Class 字段名

Phoenix 寫入完整流程

df.write
  .format("org.apache.phoenix.spark")
  .option("table", "TARGET_TABLE")
  .option("zkUrl", "zk1,zk2,zk3:2181")
  .mode(SaveMode.Overwrite)
  .save()

關鍵注意點

總結

無 Schema 無真相。 `sqlContext.createDataFrame(rdd, schema)` 或 `toDF(names)` 顯式指定列名，將 Phoenix 表結構映射到 DataFrame Schema，從源頭消除 `_1` `_2` 自動生成問題。