How do I read a Parquet file into a Spark DataFrame in PySpark?

Summary

• Use spark.read.parquet(path). It takes one or more paths and returns a DataFrame. Parquet files are self-describing, so Spark reads the schema from the files, and all columns become nullable on read.
• Select and filter early. The Databricks docs say Parquet lets query engines read only the columns needed and skip irrelevant row groups. In Spark, spark.sql.parquet.filterPushdown is true by default.
• Read the table root to get partition columns. Spark extracts key=value directory names as columns, and a filter that references only partition columns prunes the other partition directories.
• Schema merging is off by default. Set the mergeSchema option to true when files have evolved schemas. Without it, Spark picks the schema from a summary file or a random data file.
• On Databricks, read a volume path with spark.read.format("parquet").load(...). Delta Lake, which extends Parquet data files with a transaction log, is the default table format.

How do I read a Parquet file into a Spark DataFrame in PySpark?

Apache Parquet is an open source, column-oriented data file format designed for efficient data storage and retrieval. In PySpark you read it with spark.read.parquet, which loads one or more paths and returns a DataFrame.

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

df = spark.read.parquet("people.parquet")
df.printSchema()
df.show()

show() prints the first 20 rows by default. PySpark DataFrames are lazily evaluated, so transformations such as select and filter only plan the work, and an action such as collect() starts the computation.

How does Spark get the schema?

Parquet files are self-describing, so the schema is preserved. When Spark reads Parquet, all columns are automatically converted to nullable for compatibility reasons. You can also pass a schema, as a StructType or a DDL-formatted string. The PySpark docs show spark.read.schema("name string").parquet(path) to read one column. The Databricks docs recommend specifying a schema to avoid the overhead of schema inference.

How do I read only the columns and rows I need?

df = (spark.read.parquet("people.parquet")
      .select("name", "age")
      .filter("age > 30"))

The Databricks docs say Parquet allows query engines to read only the columns needed and skip irrelevant row groups. The Spark setting spark.sql.parquet.filterPushdown enables Parquet filter push-down and is true by default.

How do I read many files or a partitioned directory?

Pass several paths, for example spark.read.parquet(path1, path2). Spark's file-based input methods also accept directories and wildcards.
For a partitioned table with directories such as path/to/table/gender=male/country=US/data.parquet, pass the table root. Spark then extracts the partition columns from the paths. Their types are inferred while spark.sql.sources.partitionColumnTypeInference.enabled is true, which is the default.
Partition pruning happens in Spark's file index. In the v4.2.0 source, predicates that reference only partition columns are evaluated against the partition values, and only the matching partitions are read. A filter on an ordinary column, such as age above, is not partition pruning. It is a candidate for Parquet filter push-down instead.

How do I read files whose schemas differ?

Schema merging is a relatively expensive operation, so it is off by default. Without it, the schema is picked from a summary file or a random data file. Turn it on for one read:

merged = spark.read.option("mergeSchema", "true").parquet("data/test_table")
merged.printSchema()

You can also set spark.sql.parquet.mergeSchema to true. In the PySpark docs example, merging two DataFrames with different columns returns NULL where a file lacks a column.

Which read options help with messy directories?

Option	What it does
mergeSchema	Merges schemas collected from all Parquet part-files. Defaults to spark.sql.parquet.mergeSchema, which is false.
ignoreCorruptFiles	When true, jobs continue past corrupt files and the contents already read are still returned.
pathGlobFilter	Only includes files whose names match a glob pattern. It does not change partition discovery.
recursiveFileLookup	Loads files recursively and disables partition inferring. Defaults to false.
modifiedBefore, modifiedAfter	Include files by modification time, in YYYY-MM-DDTHH:mm:ss format.

What changes on Databricks?

• The Databricks docs read Parquet from a Unity Catalog volume path: spark.read.format("parquet").load("/Volumes/<catalog>/<schema>/<volume>/reviews_parquet"). In SQL they use read_files with format => 'parquet'.
• Delta Lake is the default format for all operations on Databricks. It extends Parquet data files with a file-based transaction log, and the docs describe a one-time conversion of Parquet tables to Delta Lake.

FAQs

Can I read Parquet from local disk on a cluster?

The Spark programming guide says that for a path on the local filesystem, the file must be accessible at the same path on worker nodes. Either copy the file to all workers or use a network-mounted shared file system.

What if a file is corrupt?

Set ignoreCorruptFiles to true, or spark.sql.files.ignoreCorruptFiles. The job continues and the contents that have been read are still returned, so check that skipping the file is acceptable.

How do I write a DataFrame back to Parquet?

Use df.write.parquet(path). DataFrameWriter.parquet also takes optional mode, partitionBy and compression arguments, for example df.write.parquet(path, mode="overwrite"). The default, error or errorifexists, throws an exception if data already exists. The other mode values are append, overwrite and ignore.

Is Parquet compressed by default?

In the Spark 4.2.0 docs, spark.sql.parquet.compression.codec defaults to snappy. The compression write option overrides it, with values such as none, uncompressed, snappy, gzip, lz4 and zstd.