如何使用Spark Scala基于每日更新文件获取最新交易记录（含删除处理）

针对你遇到的问题——无法通过分区+DenseRank移除已删除记录，以下分两种常见场景给出具体实现方案：

场景1：删除记录通过标识字段标记 #

如果更新后的文件中，被删除的记录会携带明确标识（如is_deleted=true），可按以下步骤处理：

1. 读取所有数据并提取日期
   从文件路径中解析出日期，用于后续判断记录版本：

   import org.apache.spark.sql.functions._
   
   // 读取所有分区文件（以parquet为例，CSV需添加header等参数）
   val rawDF = spark.read.parquet("transactionData/*/")
     .withColumn("file_path", input_file_name())
     // 从路径提取日期，需根据实际路径调整正则表达式
     .withColumn("record_date", regexp_extract(col("file_path"), "transactionData/(\\d{8})/", 1).cast("date"))
   

2. 筛选每个TransactionId的最新记录
   用窗口函数按TransactionId分区，按日期降序排序取最新版本：

   import org.apache.spark.sql.expressions.Window
   
   val rankedDF = rawDF
     .withColumn(
       "rank",
       dense_rank().over(Window.partitionBy("TransactionId").orderBy(col("record_date").desc))
     )
     .filter(col("rank") === 1)
     .drop("rank", "file_path", "record_date")
   

3. 过滤已删除的记录
   移除带删除标识的记录：

   val finalDF = rankedDF.filter(col("is_deleted") =!= true)
   

场景2：删除记录表现为最新日期文件中无该TransactionId #

如果删除操作是直接从最新版本文件中移除对应记录，需额外判断该ID是否在最新日期的有效记录中存在：

1. 读取数据并提取日期
   同场景1的第一步代码。

2. 获取每个TransactionId的最新出现日期

   val maxDatePerIdDF = rawDF
     .groupBy("TransactionId")
     .agg(max("record_date").alias("latest_date"))
   

3. 获取最新日期的有效TransactionId集合
   提取最新日期文件中存在的所有ID（即未被删除的记录）：

   val latestDate = rawDF.select(max("record_date")).as[java.sql.Date].first()
   val latestValidIdsDF = rawDF
     .filter(col("record_date") === latestDate)
     .select("TransactionId")
     .distinct()
   

4. 关联筛选最终有效记录
   仅保留最新日期存在的ID，并取其最新版本的记录：

   val finalDF = rawDF
     .join(maxDatePerIdDF, Seq("TransactionId"), "inner")
     .join(latestValidIdsDF, Seq("TransactionId"), "inner")
     .filter(col("record_date") === col("latest_date"))
     .drop("file_path", "record_date", "latest_date")
   

针对你的具体场景说明 #

你的场景中，20220402仅更新了xyz文件：

• abc文件的记录未变更，其latest_date为20220401，且在最新日期的有效ID集合中存在，会被保留
• xyz中修改的12345678记录，latest_date为20220402，会被保留修改后的版本
• xyz中被删除的123456789记录，不在20220402的有效ID集合中，会被排除

内容的提问来源于stack exchange，提问作者Bhagyashree Pawar