静的データベースと動的データベース（Spark SQLの小ネタ）

// S3のparquetファイルを読み込む
// 1. カラム id, date, info を有するparquetファイルが
//   「date=2020-09-14」ディレクトリの下に配置されている場合
// 2. Sparkにはpartitionという機能があり、dateカラムの日付別に
//   ディレクトリを持たせて管理することが出来ます
val dataFrame = 
  spark.read.parquet(s"s3://test-1234/sample_table/date=2020-09-14")

// 新しいカラムを追加（column_newはカラム名、次の引数でカラムの値を指定）
val dataFrameNew = 
  dataFrame.withColumn("column_new", lit("新しいカラム"))

// dataFrameNewを別のテーブルとして書き込み
dataFrameNew
  .select("id", "date", "info", "column_new")
  .write
  .option("header","false").mode("overwrite").partitionBy("date")
  .parquet(s"s3://test-1234/new_table/date=2020-09-14")

// dataFrameとdataFrameNewのスキーマを覗くとカラムが追加されたのが分かります
println(dataFrame.schema.fieldNames)
println(dataFrameNew.schema.fieldNames)

前提としてスキーマ情報は以下とします。
  id　数値
  date　文字列(yyyy-mm-dd)
  info_a　文字列
  info_b　文字列

■テーブル１
s3://test-1234/sample_table/date=2020-09-15/aaaa.parquet
  => カラムは id, date, info_a を含む
  +----+----------+------+
  | id | date     |info_a|
  +----+----------+------+
  |   1|2020-09-15|   red|
  |   2|2020-09-15|  blue|
  +----+----------+------+

■テーブル２
s3://test-1234/sample_table/date=2020-09-16/bbbb.parquet
  => カラムは id, date, info_b を含む
  +----+----------+------+
  | id | date     |info_b|
  +----+----------+------+
  |   1|2020-09-16| green|
  |   2|2020-09-16|  gray|
  +----+----------+------+


上記の２つのテーブルを結合して以下のようなテーブルを生成したい
+----+----------+------+------+                                                       
| id | date     |info_a|info_b|
+----+----------+------+------+
|   1|2020-09-15|   red|  null|
|   2|2020-09-15|  blue|  null|
|   1|2020-09-16|  null| green|
|   2|2020-09-16|  null|  gray|
+----+----------+------+------+

val dateList = List("2020-09-15", "2020-09-16")

// date partitionごとにファイルを読み込んでDataFrameのリストを生成
val dfListByDate = 
  // 日付のリストをDataFrameのリストに変換する
  dateList.map { date => 
    spark.read
      .parquet(s"s3://test-1234/sample/date=$date")
      // partitionごとにreadするとdateカラムが消えてしまうので付加しなおす
      .withColumn("date", lit(s"$date")) 
  }

// 最終的に必要となるカラム名のリストを作成
// 今回の例の場合、List("id", "date", "info_a", "info_b")が得られる
val mergedFieldNames = 
  dfListByDate.map(_.schema.fieldNames).flatten.toSet.toList

// 最後にDataFrameのお互いに足りないカラムをNULL値で補い結合します。
// ここでポイントとなるのはselect()する際に個別のカラム名ではなく
// カラム名のリストを渡すことです。Spark SDKの素晴らしい柔軟性ですね。
val result = 
  // カラムの追加後、reduceLeft()で全てのDataFrameを結合
  dfListByDate.map { dfByDate => 
    // カラム名のリストのdiffを取れば不足しているカラム名が得られる
    val neededfieldNames = 
      mergedFieldNames diff dfByDate.schema.fieldNames;

    // 不足しているカラムがない場合はselect()して順番を合わせるだけ
    if (neededfieldNames.length == 0) {
      dfByDate.select(mergedFieldNames.map(col): _*) 
    // 不足しているカラムがある場合はwithColumn()で付加してからselect()する
    } else { 
      neededfieldNames.foldLeft(dfByDate) {
        (df: Dataset[Row], fieldName) => 
          df.withColumn(fieldName, lit(null)) 
      }.select(mergedFieldNames.map(col): _*) 
    } 
  }.reduceLeft(_ union _)

// 結合完了！！！！
result.show