PySpark操作Parquet：忽略缺失值及数组转结构体数组需求

Hey there! Let's work through your two PySpark Parquet challenges step by step—both are pretty common when dealing with nested data, so I’ve got practical solutions for you.

需求一：写入Parquet时忽略缺失值 #

The key here is to clean up missing values before writing to Parquet, since PySpark doesn’t have a built-in "ignore nulls on write" flag. The dropna() method is your go-to tool here:

• To remove any row that has a null value in any column:

# Assume your DataFrame is named `raw_df`
cleaned_df = raw_df.dropna()
# Write the cleaned data to Parquet
cleaned_df.write.parquet("/path/to/your/cleaned_data.parquet")

• If you only want to drop rows where nulls appear in specific columns, use the subset parameter:

# Only drop rows where "critical_col1" or "critical_col2" have nulls
cleaned_df = raw_df.dropna(subset=["critical_col1", "critical_col2"])
cleaned_df.write.parquet("/path/to/your/cleaned_data.parquet")

If deleting rows isn’t an option, you can also fill nulls with default values using fillna() instead:

# Fill numeric nulls with 0.0 and string nulls with "unknown"
filled_df = raw_df.fillna({"numeric_col": 0.0, "string_col": "unknown"})
filled_df.write.parquet("/path/to/your/filled_data.parquet")

需求二：将Array[Array[Float]]转换为Array[Struct] #

You’re already on the right track by defining the target struct schema! Now we just need to map each inner array to a struct using PySpark’s higher-order transform() function.

Let’s say your source column is called nested_arrays, where each inner array has exactly 4 float values matching your struct fields ("one", "two", "three", "four"). Here’s how to do the conversion:

from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType, StructType, StructField, FloatType

# Define your target struct schema
target_struct = StructType([
    StructField("one", FloatType()),
    StructField("two", FloatType()),
    StructField("three", FloatType()),
    StructField("four", FloatType())
])

# Transform the nested array column to array of structs
transformed_df = raw_df.withColumn(
    "struct_array",
    F.transform(
        "nested_arrays",
        lambda inner_arr: F.struct(
            inner_arr[0].alias("one"),
            inner_arr[1].alias("two"),
            inner_arr[2].alias("three"),
            inner_arr[3].alias("four")
        )
    )
)

# Optional: Replace the original column if needed
transformed_df = transformed_df.drop("nested_arrays").withColumnRenamed("struct_array", "nested_arrays")

# Write the transformed data to Parquet
transformed_df.write.parquet("/path/to/your/transformed_data.parquet")

Pro Tip: Handle inconsistent inner array lengths #

If some inner arrays don’t have exactly 4 elements, you’ll run into index errors. Add a filter first to keep only valid inner arrays:

# Filter out inner arrays that don't have 4 elements
filtered_df = raw_df.withColumn(
    "nested_arrays",
    F.filter("nested_arrays", lambda arr: F.size(arr) == 4)
)

# Then run the transform step as above

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