PySpark中基于日期与URL模糊匹配的DataFrame连接实现问询

Hey there! Let's convert your SQL inner join logic to PySpark DataFrame API smoothly. I'll walk you through the exact code and some performance tips since you're dealing with a massive 1-billion-row dataset—efficiency matters here.

First, let's recap your original SQL logic to make sure we cover every condition:

• Inner join between link_data_test (aliased as t1) and url_data_test (aliased as t2)
• t1.link_date falls strictly between t2.extra_date and t2.extra_date + 8
• t1.tuch_url contains the value of t2.url (equivalent to t1.tuch_url LIKE '%t2.url%' in SQL)

Here's the step-by-step implementation with DataFrame API:

Step 1: Import required functions #

You'll need a few PySpark functions to replicate the date calculation and LIKE pattern:

from pyspark.sql.functions import date_add, concat, lit

Step 2: Alias your DataFrames #

Using aliases keeps the condition code clean and matches your original SQL table references:

t1 = A_dataframe.alias("t1")
t2 = B_dataframe.alias("t2")

Step 3: Build the join condition #

Translate each SQL condition into DataFrame expressions, combining them with & (logical AND):

join_condition = (
    (t1.link_date > t2.extra_date) &
    (t1.link_date < date_add(t2.extra_date, 8)) &
    (t1.tuch_url.like(concat(lit('%'), t2.url, lit('%'))))
)

• The like method uses concat(lit('%'), t2.url, lit('%')) to replicate the '%t2.url%' wildcard pattern from your SQL.

Step 4: Execute the inner join #

While inner join is the default behavior, we'll explicitly set how="inner" for clarity:

joined_df = t1.join(t2, join_condition, how="inner")

Step 5: Save the result to a table #

Just like your original SQL workflow, you can persist the joined DataFrame to a table:

# Replace "your_target_table_name" with your actual desired table name
joined_df.write.saveAsTable("your_target_table_name")

Performance Tips for Large Datasets #

Since A_dataframe has 1 billion rows, here are a few optimizations to speed up the join:

• Partition strategically: Ensure link_date and extra_date are used for partitioning where possible—this reduces the amount of data shuffled during the join.
• Broadcast small tables: If B_dataframe is significantly smaller than A_dataframe, use the broadcast function to send it to all worker nodes, avoiding expensive shuffles:

  from pyspark.sql.functions import broadcast
  joined_df = t1.join(broadcast(t2), join_condition, how="inner")
  

• Deduplicate first: If B_dataframe has duplicate url values, deduplicate it with t2 = B_dataframe.dropDuplicates(["url"]) to cut down on unnecessary join pairs.

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