변경 데이터 캡처로 업스트림 데이터 변경 처리

+---+-----+
| id| name|
+---+-----+
|id1|Alice|
|id2|  Bob|
+---+-----+

case class ChangeEvent(
  changeType: String,           // type of change - INSERT, UPDATE or DELETE
  timestamp: String,            // when this change happened 
  columnNames: Seq[String],     // names of columns (Mandatory for INSERT / UPDATE)
  columnValues: Seq[String],    // values of columns (Mandatory for INSERT / UPDATE)
  oldKeyNames: Seq[String],     // names of old key columns (Mandatory for UPDATE / DELETE)
  oldKeyValues: Seq[String]     // values of old key columns (Mandatory for UPDATE / DELETE)
)

import org.apache.spark.sql.{DataFrame, Row}
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.expressions.Window

val schema = StructType(Seq("timestamp", "id", "name").map(StructField(_, StringType)))
val datalakeDf = spark.createDataFrame(spark.sparkContext.parallelize(Seq(
    Row("0", "id1", "Alice"),
    Row("0", "id2", "Bob")
)), schema)

+---------+---+-----+
|timestamp| id| name|
+---------+---+-----+
|        0|id1|Alice|
|        0|id2|  Bob|
+---------+---+-----+

val COLUMN_NAMES = Seq("id", "name")  // column names of the data
val OLD_KEY_NAMES = Seq("id")         // column names of the primary key

val changeEventDf = spark.createDataFrame(Seq(
    ChangeEvent("update", "1", COLUMN_NAMES, Seq("id1", "Angela"), OLD_KEY_NAMES, Seq("id1")),
    ChangeEvent("delete", "2", null, null, OLD_KEY_NAMES, Seq("id2")),
    ChangeEvent("insert", "3", COLUMN_NAMES, Seq("id2", "Carol"), null, null)
))

+----------+---------+-----------+-------------+-----------+------------+
|changeType|timestamp|columnNames| columnValues|oldKeyNames|oldKeyValues|
+----------+---------+-----------+-------------+-----------+------------+
|    update|        1| [id, name]|[id1, Angela]|       [id]|       [id1]|
|    delete|        2|       null|         null|       [id]|       [id2]|
|    insert|        3| [id, name]| [id2, Carol]|       null|        null|
+----------+---------+-----------+-------------+-----------+------------+

val splitChangeEventDf = COLUMN_NAMES.zipWithIndex.foldLeft(changeEventDf) {
    (df, column) => df.withColumn(column._1, $"columnValues"(column._2))
}.drop("columnNames", "columnValues")

+----------+---------+-----------+------------+----+------+
|changeType|timestamp|oldKeyNames|oldKeyValues|  id|  name|
+----------+---------+-----------+------------+----+------+
|    update|        1|       [id]|       [id1]| id1|Angela|
|    delete|        2|       [id]|       [id2]|null|  null|
|    insert|        3|       null|        null| id2| Carol|
+----------+---------+-----------+------------+----+------+

  def unionWithSchema(dataFrames: DataFrame *): Option[DataFrame] = {

    if(dataFrames.isEmpty) {
      return None
    }
    val spark = dataFrames.head.sparkSession
    val distinctSchemas = dataFrames.map(_.schema.toList).distinct
    val unionDf = if(distinctSchemas.size == 1) {
      dataFrames.tail.foldLeft(dataFrames.head) {
        (df1, df2) => df1.union(df2)
      }
    } else {
      val allSchemas = distinctSchemas.flatten.distinct.sortBy(schema => schema.name)
      val schemaWithAllColumns = StructType(allSchemas)
      val emptyDataFrame = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], schemaWithAllColumns)
      val orderedColumns = emptyDataFrame.columns.map(col)

      dataFrames.foldLeft(emptyDataFrame) {
        (df1, df2) => {
          val missingColumns = allSchemas diff df2.schema
          val unionSafeDf = missingColumns.foldLeft(df2) {
            (df, missingField) => df.withColumn(missingField.name, lit(null).cast(missingField.dataType))
          }
          df1.union(unionSafeDf.select(orderedColumns: _*))
        }
      }
    }
    Some(unionDf)
}

  def applyChangeEventsByTimestamp(dataFrame: DataFrame, primaryKeyArr: Array[String]): DataFrame = {

    val partitionCols = if(primaryKeyArr.isEmpty) {
      dataFrame.columns.map(col)
    } else {
      primaryKeyArr.map(colName => col(s"old_$colName"))
    }
    val window = Window.partitionBy(partitionCols: _*).orderBy(desc("timestamp"))

    var dfInitial = dataFrame
    var iteration = 0
    var done = false

    // Potentially expensive. Monitor logs and improve if required.
    while (!done) {
      val dfBefore = if (iteration == 0) {
        dfInitial
      } else {
        primaryKeyArr.foldLeft(dfInitial) {
          // Update the old keys for UPDATE ops
          (df, colName) => df.withColumn(s"old_$colName",
            when($"changeType" === "update", col(colName)).otherwise(col(s"old_$colName")))
        }
      }

      val dfAfter = dfBefore.withColumn("rank", row_number().over(window))
        .filter(col("rank") === 1)
        .drop("rank")

      done = dfAfter.count == dfBefore.count
      dfInitial = dfAfter
      iteration = iteration + 1
    }
    dfInitial
}

val insertDf = splitChangeEventDf.filter($"changeType" === "insert").drop("oldKeyNames", "oldKeyValues")
val datalakeAndInsertDf = unionWithSchema(insertDf, datalakeDf).get

+----------+---+-----+---------+
|changeType| id| name|timestamp|
+----------+---+-----+---------+
|    insert|id2|Carol|        3|
|      null|id1|Alice|        0|
|      null|id2|  Bob|        0|
+----------+---+-----+---------+

val updateDeleteDf = splitChangeEventDf.filter($"changeType" === "update" || $"changeType" === "delete")
val distinctOldKeyNames = updateDeleteDf.select("oldKeyNames").distinct.collect.flatMap(_.getSeq[String](0))

// Ideally there should only be one but primary key can be changed in some cases
val updateDeleteExpandedDfs = distinctOldKeyNames.foldLeft(datalakeAndInsertDf)(

    (datalakeAndInsertDf, primaryKeyArr) => {

        // Step 1: For INSERT / Existing data, create new pkey columns (using existing values)
        val datalakeAndInsertCeDf = primaryKeyArr.foldLeft(datalakeAndInsertDf) {
            (df, colName) => df.withColumn(s"old_$colName", col(colName))
        }

        // Step 2: For UPDATE / DELETE, split the old keys array column
        val updateDeleteCeDf = primaryKeyArr.zipWithIndex.foldLeft(updateDeleteDf) {
            (df, pKey) => df.withColumn(s"old_${pKey._1}", $"oldKeyValues"(pKey._2))
        }
        .filter($"oldKeyNames" === primaryKeyArr)
        .drop("oldKeyNames", "oldKeyValues")

        // Step 3: Union all the data
        val initialDf = unionWithSchema(datalakeAndInsertCeDf, updateDeleteCeDf).get.cache()

        // Step 4: Resolve the change events chronologically
        val resolvedDf = applyChangeEventsByTimestamp(initialDf, primaryKeyArr)

        // Step 5: Drop DELETE records and unnecessary columns
        resolvedDf
            .filter($"changeType" =!= "delete" || $"changeType".isNull)
            .drop(primaryKeyArr.map(colName => s"old_$colName"):  _*)
    }
}
// Step 6: Remove unwanted columns and get rid of duplicate rows
return updateDeleteExpandedDfs.drop("changeType").distinct()

scala> datalakeAndInsertCeDf.show
+----------+---+-----+---------+------+
|changeType| id| name|timestamp|old_id|
+----------+---+-----+---------+------+
|    insert|id2|Carol|        3|   id2|
|      null|id1|Alice|        0|   id1|
|      null|id2|  Bob|        0|   id2|
+----------+---+-----+---------+------+

scala> updateDeleteCeDf.show
+----------+---------+----+------+------+
|changeType|timestamp|  id|  name|old_id|
+----------+---------+----+------+------+
|    update|        1| id1|Angela|   id1|
|    delete|        2|null|  null|   id2|
+----------+---------+----+------+------+

scala> initialDf.show
+----------+----+------+------+---------+
|changeType|  id|  name|old_id|timestamp|
+----------+----+------+------+---------+
|    insert| id2| Carol|   id2|        3|
|      null| id1| Alice|   id1|        0|
|      null| id2|   Bob|   id2|        0|
|    update| id1|Angela|   id1|        1|
|    delete|null|  null|   id2|        2|
+----------+----+------+------+---------+

scala> resolvedDf.show
+----------+---+------+------+---------+
|changeType| id|  name|old_id|timestamp|
+----------+---+------+------+---------+
|    update|id1|Angela|   id1|        1|
|    insert|id2| Carol|   id2|        3|
+----------+---+------+------+---------+

+----------+---+------+---------+
|changeType| id|  name|timestamp|
+----------+---+------+---------+
|    update|id1|Angela|        1|
|    insert|id2| Carol|        3|
+----------+---+------+---------+

+---+------+---------+
| id|  name|timestamp|
+---+------+---------+
|id2| Carol|        3|
|id1|Angela|        1|
+---+------+---------+