Spark避免临时视图：SQL逻辑转DataFrame API实现咨询

Got it, let's walk through converting your temp view-based SQL logic to pure DataFrame operations step by step. We'll replicate the window function ranking and join behavior without relying on temporary views.

Key Concepts to Replicate #

• Join table1 with table2, then use a rank window function to keep only the latest table2 records per group
• Join that result with table3, repeat the rank logic to get the latest table3 records
• Write the final result to your target table

Python Implementation #

First, import the necessary Spark functions:

from pyspark.sql import Window
from pyspark.sql.functions import col, rank

Step 1: Join table1 and table2, retain latest table2 records

Define a window specification partitioned by your join key(s), ordered by the timestamp column (used to determine "latest") in descending order. Then add a rank column and filter to keep only the top-ranked (latest) records:

# Define window for table2: partition by your join key, order by timestamp descending
window_table2 = Window.partitionBy("your_join_key_col").orderBy(col("table2_timestamp_col").desc())

# Join tables, add rank, filter for latest records
table1_table2_df = table1.join(table2, on="your_join_key_col", how="inner")  # adjust join type (inner/left/etc.) to match your SQL
    .withColumn("table2_rank", rank().over(window_table2))
    .filter(col("table2_rank") == 1)
    .drop("table2_rank", "table2_timestamp_col")  # clean up unused columns

Step 2: Join with table3, retain latest table3 records

Repeat the window function pattern for the table3 join:

# Define window for table3
window_table3 = Window.partitionBy("another_join_key_col").orderBy(col("table3_timestamp_col").desc())

# Join and filter for latest table3 records
final_df = table1_table2_df.join(table3, on="another_join_key_col", how="inner")
    .withColumn("table3_rank", rank().over(window_table3))
    .filter(col("table3_rank") == 1)
    .drop("table3_rank", "table3_timestamp_col")

Step 3: Write to target table

# Adjust write mode (overwrite/append/etc.) as needed
final_df.write.mode("overwrite").saveAsTable("your_target_table")

Scala Implementation #

If you're using Scala, the logic is identical—just syntax changes:

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{col, rank}

// Step 1: Join table1 & table2, get latest table2 records
val windowTable2 = Window.partitionBy("your_join_key_col").orderBy(col("table2_timestamp_col").desc)

val table1Table2Df = table1.join(table2, Seq("your_join_key_col"), "inner")
    .withColumn("table2_rank", rank().over(windowTable2))
    .filter(col("table2_rank") === 1)
    .drop("table2_rank", "table2_timestamp_col")

// Step 2: Join with table3, get latest table3 records
val windowTable3 = Window.partitionBy("another_join_key_col").orderBy(col("table3_timestamp_col").desc)

val finalDf = table1Table2Df.join(table3, Seq("another_join_key_col"), "inner")
    .withColumn("table3_rank", rank().over(windowTable3))
    .filter(col("table3_rank") === 1)
    .drop("table3_rank", "table3_timestamp_col")

// Step 3: Save to target table
finalDf.write.mode("overwrite").saveAsTable("your_target_table")

Important Notes #

• Replace placeholders: Swap out columns like your_join_key_col, table2_timestamp_col with your actual schema fields
• Join types: Adjust how="inner" to match your original SQL (e.g., left, right, full)
• Rank vs Row Number: If you want to handle ties (multiple records with the same latest timestamp) by picking exactly one, use row_number() instead of rank()
• Additional transformations: Add any filters, column renames, or calculations from your original SQL using .filter(), .withColumn(), or .select() as needed

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