如何在PySpark中转换嵌套DataFrame的Schema结构

Got it, let's walk through a concrete, practical example to show how to convert a nested PySpark DataFrame from one schema to another. I'll start by defining a realistic source schema, a target schema we want to map to, and then the transformation code to make it happen.

Source Nested Schema #

Let's use a common e-commerce-style nested structure with user details, nested orders, and nested line items inside each order:

from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType, ArrayType, DoubleType
)

source_schema = StructType([
    StructField("user_id", StringType(), nullable=False),
    StructField("user_name", StringType(), nullable=False),
    StructField("contact_info", StructType([
        StructField("email", StringType(), nullable=True),
        StructField("phone", StringType(), nullable=True)
    ])),
    StructField("orders", ArrayType(StructType([
        StructField("order_id", StringType(), nullable=False),
        StructField("order_date", StringType(), nullable=False),
        StructField("shipping_address", StructType([
            StructField("street", StringType(), nullable=False),
            StructField("city", StringType(), nullable=False)
        ])),
        StructField("line_items", ArrayType(StructType([
            StructField("product_id", StringType(), nullable=False),
            StructField("quantity", IntegerType(), nullable=False),
            StructField("unit_price", DoubleType(), nullable=False)
        ])))
    ])))
])

Sample source data snippet to visualize the structure:

{
"user_id": "U1001",
"user_name": "Jane Smith",
"contact_info": {"email": "jane@example.com", "phone": "555-1234"},
"orders": [
{
"order_id": "O2001",
"order_date": "2024-02-20",
"shipping_address": {"street": "123 Main St", "city": "New York"},
"line_items": [
{"product_id": "P3001", "quantity": 2, "unit_price": 19.99},
{"product_id": "P3002", "quantity": 1, "unit_price": 49.99}
]
}
]
}

Target Schema #

Let's say we want to flatten the structure, move order and line item fields to the top level, and add a calculated line_item_total field (using the formula quantity * unit_price):

target_schema = StructType([
    StructField("user_id", StringType(), nullable=False),
    StructField("user_name", StringType(), nullable=False),
    StructField("user_email", StringType(), nullable=True),
    StructField("user_phone", StringType(), nullable=True),
    StructField("order_id", StringType(), nullable=False),
    StructField("order_date", StringType(), nullable=False),
    StructField("shipping_city", StringType(), nullable=False),
    StructField("product_id", StringType(), nullable=False),
    StructField("quantity", IntegerType(), nullable=False),
    StructField("unit_price", DoubleType(), nullable=False),
    StructField("line_item_total", DoubleType(), nullable=False)
])

Transformation Code #

Here's how to convert the source DataFrame to match the target schema using PySpark's built-in functions:

from pyspark.sql import functions as F

# Assume `source_df` is your DataFrame loaded with the source_schema
transformed_df = (
    source_df
    # Flatten the contact_info struct first
    .withColumn("user_email", F.col("contact_info.email"))
    .withColumn("user_phone", F.col("contact_info.phone"))
    # Explode the orders array to get one row per order
    .withColumn("order", F.explode(F.col("orders")))
    # Explode line items to get one row per product in the order
    .withColumn("line_item", F.explode(F.col("order.line_items")))
    # Select and reshape all fields to match the target schema
    .select(
        "user_id",
        "user_name",
        "user_email",
        "user_phone",
        F.col("order.order_id"),
        F.col("order.order_date"),
        F.col("order.shipping_address.city").alias("shipping_city"),
        F.col("line_item.product_id"),
        F.col("line_item.quantity"),
        F.col("line_item.unit_price"),
        # Calculate derived field using the formula
        (F.col("line_item.quantity") * F.col("line_item.unit_price")).alias("line_item_total")
    )
)

# Verify the output schema matches the target
transformed_df.printSchema()

Key Tips for Schema Conversion: #

• Use F.explode() to flatten array fields into individual rows
• Access nested struct fields with dot notation (e.g., contact_info.email)
• Use alias() to rename fields to match the target schema's naming conventions
• Calculate derived fields directly in the select clause using PySpark's arithmetic functions

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