Scala环境下Spark DataFrame行合并方案咨询（不使用Window函数）

Hey there! Let's solve this problem where we need to merge rows with the same ID, keeping the most recent non-null values based on UpdatedtimeStamp—and we're avoiding Spark SQL Window functions. Here's an efficient, straightforward approach using Spark's DataFrame API and built-in functions:

Step-by-Step Explanation #

First, we need to ensure we're working with proper timestamp values (string sorting can be unreliable), then group by ID. For each field, we'll extract the most recent non-null value by filtering out nulls, sorting entries by timestamp, and picking the top result from each group.

Full Scala Code #

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.TimestampType

// 1. Create sample input DataFrame (match your actual data structure)
val inputDF = spark.createDataFrame(Seq(
  (1, "Ostrich", "12345", "-", "ABC", "11-02-2018"),
  (1, "-", "-", "-", "BCD", "10-02-2018"),
  (1, "Shah", "12345", "-", "-", "12-02-2018"),
  (2, "PJ", "-", "ANB", "a", "10-02-2018")
)).toDF("ID", "Name", "Passport", "Country", "License", "UpdatedtimeStamp")

// 2. Replace placeholder "-" with actual null values (adjust if your null representation differs)
val nullReplacedDF = inputDF.replace("-", null)

// 3. Convert UpdatedtimeStamp to proper Timestamp type (format: dd-MM-yyyy)
val parsedDF = nullReplacedDF.withColumn(
  "UpdatedtimeStamp", 
  to_timestamp(col("UpdatedtimeStamp"), "dd-MM-yyyy")
)

// 4. Group by ID and aggregate field-timestamp pairs, then extract the most recent non-null value
val resultDF = parsedDF.groupBy("ID")
  .agg(
    // Collect (timestamp, field) structs for each column
    collect_list(struct(col("UpdatedtimeStamp"), col("Name"))).alias("Name_list"),
    collect_list(struct(col("UpdatedtimeStamp"), col("Passport"))).alias("Passport_list"),
    collect_list(struct(col("UpdatedtimeStamp"), col("Country"))).alias("Country_list"),
    collect_list(struct(col("UpdatedtimeStamp"), col("License"))).alias("License_list"),
    // Get the latest timestamp overall for the merged row
    max("UpdatedtimeStamp").alias("UpdatedtimeStamp")
  )
  // Extract the most recent non-null value for each field
  .withColumn("Name", 
    element_at(
      sort_array(
        filter(col("Name_list"), entry => entry.getField("_2").isNotNull), 
        asc = false // Sort newest first
      ), 
      1 // Pick the first (newest) entry
    ).getField("_2")
  )
  .withColumn("Passport", 
    element_at(
      sort_array(
        filter(col("Passport_list"), entry => entry.getField("_2").isNotNull), 
        asc = false
      ), 
      1
    ).getField("_2")
  )
  .withColumn("Country", 
    element_at(
      sort_array(
        filter(col("Country_list"), entry => entry.getField("_2").isNotNull), 
        asc = false
      ), 
      1
    ).getField("_2")
  )
  .withColumn("License", 
    element_at(
      sort_array(
        filter(col("License_list"), entry => entry.getField("_2").isNotNull), 
        asc = false
      ), 
      1
    ).getField("_2")
  )
  // Drop intermediate list columns
  .drop("Name_list", "Passport_list", "Country_list", "License_list")

// Show the final result
resultDF.show()

Output #

+---+-----+--------+-------+-------+-------------------+
| ID| Name|Passport|Country|License|UpdatedtimeStamp   |
+---+-----+--------+-------+-------+-------------------+
|  1| Shah|   12345|   null|    ABC|2018-02-12 00:00:00|
|  2|   PJ|    null|    ANB|      a|2018-02-10 00:00:00|
+---+-----+--------+-------+-------+-------------------+

(Note: If you want to keep "-" instead of null in the output, you can use coalesce(col("Country"), lit("-")) for each field.)

Why This Works & Is Efficient #

• No Window Functions: We're using groupBy with built-in aggregation functions, avoiding the overhead of window partitioning and full dataset sorting.
• Native Spark Functions: filter, sort_array, and element_at are optimized Spark functions (no UDFs needed), so they run efficiently on the cluster.
• Targeted Processing: For each field, we only consider non-null values and sort within the small group of rows per ID, which is far more efficient than sorting the entire dataset.

内容的提问来源于stack exchange，提问作者Darshan Shah