PySpark写入Parquet至HDFS时，如何生成同目录元数据/Schema文件？

Great questions! Let's walk through how to generate standalone metadata/schema files alongside your Parquet outputs in PySpark, whether you're writing a single DataFrame or combining multiple Parquet sources first.

1. Creating a metadata file when writing a DataFrame to Parquet on HDFS #

Parquet files already embed their schema, but if you need a separate, easily accessible metadata file (like a JSON schema), here's how to do it:

First, write your DataFrame to Parquet as usual:

# Define your output path on HDFS
output_path = "hdfs://your/cluster/path/your_parquet_data"

# Write the DataFrame (use mode="overwrite" if you want to replace existing data)
your_dataframe.write.mode("overwrite").parquet(output_path)

Next, extract the DataFrame's schema as a JSON string and write it to the same directory. You have two straightforward options:

Option 1: Use Spark's DataFrame writer (simple, no extra imports) #

# Get the schema as a formatted JSON string
schema_json = your_dataframe.schema.json()

# Create a single-row DataFrame with the schema string
schema_df = spark.createDataFrame([(schema_json,)], ["schema"])

# Write the schema to a text file in the same output directory
# Using .text() ensures the output is a plain JSON string (easy to read/parse)
schema_df.select("schema").write.mode("overwrite").text(f"{output_path}/_schema.json")

Option 2: Use Hadoop FileSystem API (more efficient, avoids creating a DataFrame) #

If you want to skip creating a temporary DataFrame, you can write directly to HDFS using the underlying Hadoop API:

from py4j.java_gateway import java_import
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# Import Hadoop FS classes
java_import(spark._jvm, "org.apache.hadoop.fs.FileSystem")
java_import(spark._jvm, "org.apache.hadoop.fs.Path")

# Get the HDFS filesystem instance
fs = spark._jvm.FileSystem.get(spark._jsc.hadoopConfiguration())

# Define the path for your schema file
schema_file_path = spark._jvm.Path(f"{output_path}/_schema.json")

# Write the schema JSON directly to the file
with fs.create(schema_file_path) as output_stream:
    output_stream.write(your_dataframe.schema.json().encode("utf-8"))

2. Generating a schema file for a combined DataFrame from two Parquet sources #

The process here is almost identical—you just add your transformation step before writing the final DataFrame. Here's a complete example:

# Step 1: Read the two input Parquet files
df1 = spark.read.parquet("hdfs://path/to/first/input_parquet")
df2 = spark.read.parquet("hdfs://path/to/second/input_parquet")

# Step 2: Perform your transformation (adjust this to your actual logic)
# Example: Join the two DataFrames on a common key
final_df = df1.join(df2, on="user_id", how="inner")

# Step 3: Write the final DataFrame to Parquet
final_output_path = "hdfs://path/to/final/output_parquet"
final_df.write.mode("overwrite").parquet(final_output_path)

# Step 4: Write the schema file to the same directory (using Option 1 from above)
final_schema_json = final_df.schema.json()
schema_final_df = spark.createDataFrame([(final_schema_json,)], ["schema"])
schema_final_df.select("schema").write.mode("overwrite").text(f"{final_output_path}/_schema.json")

Key Notes: #

• The JSON schema file will be compatible with most tools (like Pandas, DBT, or custom scripts) that need to parse the DataFrame structure without reading the entire Parquet dataset.
• If you prefer a different format (like YAML), you can convert the JSON schema to YAML using libraries like pyyaml (just make sure to install it on your Spark workers if you're running distributed).
• Always use the same mode (e.g., overwrite) for both the Parquet write and schema write to ensure consistency if you're replacing existing data.

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