lateralJoin

使用给定联接表达式将横向联接与另一个 DataFrame 联接。

Syntax

lateralJoin(other: "DataFrame", on: Optional[Column] = None, how: Optional[str] = None)

参数

退货

DataFrame：联接的数据帧。

备注

横向联接（也称为关联联接）是一种联接类型，其中一个 DataFrame 中的每一行用作子查询或派生表的输入，用于计算特定于该行的结果。 右侧 DataFrame 可以从左侧 DataFrame的当前行引用列，从而比标准联接更复杂且依赖于上下文的结果。

示例

from pyspark.sql import functions as sf
from pyspark.sql import Row
customers_data = [
    Row(customer_id=1, name="Alice"), Row(customer_id=2, name="Bob"),
    Row(customer_id=3, name="Charlie"), Row(customer_id=4, name="Diana")
]
customers = spark.createDataFrame(customers_data)
orders_data = [
    Row(order_id=101, customer_id=1, order_date="2024-01-10",
        items=[Row(product="laptop", quantity=5), Row(product="mouse", quantity=12)]),
    Row(order_id=102, customer_id=1, order_date="2024-02-15",
        items=[Row(product="phone", quantity=2), Row(product="charger", quantity=15)]),
    Row(order_id=105, customer_id=1, order_date="2024-03-20",
        items=[Row(product="tablet", quantity=4)]),
    Row(order_id=103, customer_id=2, order_date="2024-01-12",
        items=[Row(product="tablet", quantity=8)]),
    Row(order_id=104, customer_id=2, order_date="2024-03-05",
        items=[Row(product="laptop", quantity=7)]),
    Row(order_id=106, customer_id=3, order_date="2024-04-05",
        items=[Row(product="monitor", quantity=1)]),
]
orders = spark.createDataFrame(orders_data)

customers.join(orders, "customer_id").lateralJoin(
    spark.tvf.explode(sf.col("items").outer()).select("col.*")
).select(
    "customer_id", "name", "order_id", "order_date", "product", "quantity"
).orderBy("customer_id", "order_id", "product").show()
# +-----------+-------+--------+----------+-------+--------+
# |customer_id|   name|order_id|order_date|product|quantity|
# +-----------+-------+--------+----------+-------+--------+
# |          1|  Alice|     101|2024-01-10| laptop|       5|
# |          1|  Alice|     101|2024-01-10|  mouse|      12|
# ...
# +-----------+-------+--------+----------+-------+--------+