Reading CSV, JSON, and Parquet Files

Spark tutorial · PySpark.in

Reading CSV, JSON, and Parquet Files

Spark can directly read CSV, JSON, and Parquet files into DataFrames using spark.read. For example, spark.read.csv("file.csv"), spark.read.json("file.json"), and spark.read.parquet("file.parquet") will load the data into DataFrames. By default, CSV is expected to be comma-separated (with one record per line) and JSON is expected to be newline-delimited JSON (one JSON object per line). Parquet files are self-describing columnar files (schema is preserved automatically).

Example (CSV): Suppose people.csv has a header row. We can read it as follows:

```

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ReadCSV").getOrCreate()

# Read a CSV file with header and inferSchema

df = spark.read.option("header", True).option("inferSchema", True).csv("s3a://pyspark420/pyspark-upload/people.csv ")

df.show()

```

This will display the DataFrame content with properly parsed columns. You can adjust options like delimiter (default is comma) or set header=False if there is no header.

Example (JSON): For a JSON file people.json (one object per line), simply do:

```

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ReadCSV").getOrCreate()

df = spark.read. option("multiline", "true").json("s3a://pyspark420/pyspark-upload/people.json")

df.printSchema()

df.show()

```

Spark will infer the schema automatically and load the JSON objects into columns.

Example (Parquet): Reading Parquet is even simpler because Parquet files include schema:

```

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ReadCSV").getOrCreate()

df = spark.read.parquet("s3a://pyspark420/pyspark-upload/people.parquet/")

df.show()

```

This returns a DataFrame with the original schema. Parquet reading automatically merges schema information and is very fast.

Advanced: You can read multiple files or directories by passing a list of paths. For example, reading two JSON directories:

```

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ExportPeople").getOrCreate()

# Read two JSON files into one DataFrame

df = spark.read.json(["s3a://pyspark420/pyspark-upload/2020.json", "s3a://pyspark420/pyspark-upload/2021.json"])

df.show(truncate=False)

```

You can also provide your own schema to avoid a costly schema inference pass:

```

from pyspark.sql import SparkSession

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

spark = SparkSession.builder.appName("ExportPeople").getOrCreate()

schema = StructType([StructField("name", StringType()), StructField("age", IntegerType())])

df = spark.read.schema(schema).json("s3a://pyspark420/pyspark-upload/people.json")

df.show()

```

For Parquet, you can enable schema merging if different files have different columns:

```

hide

df = spark.read.option("mergeSchema", "true").parquet("dir1/", "dir2/")

```

This merges columns from all files.

Writing CSV, JSON, and Parquet Files

DataFrames can be saved back to CSV, JSON, or Parquet using df.write. For example, df.write.csv("out.csv"), df.write.json("out.json"), and df.write.parquet("out.parquet") will write the DataFrame to files. By default, Spark writes in a directory (it creates part files and a _SUCCESS file).

Example (CSV): To write a DataFrame to CSV with a header row:

```

hide

# Using the DataFrame from before:

df.write.mode("overwrite").option("header", True).csv("output/people_csv")

```

This creates a folder output/people_csv/ containing CSV part-files. Using mode("overwrite") ensures any existing data is replaced. The header=True option writes the column names as the first line.

Example (JSON): To save as JSON lines:

```

hide

df.write.mode("overwrite").json("output/people_json")

```

This writes each row as a JSON object (newline-delimited) in the output directory.

Example (Parquet): To write as Parquet:

```

hide

df.write.mode("overwrite").parquet("output/people_parquet")

```

Parquet automatically preserves the schema. The output directory will contain Parquet files. By default, Spark uses the “snappy” codec for Parquet (see Compression below).

Advanced: You can combine writing with other options. For example, writing CSV with a custom delimiter and compression:

```

hide

df.write

df.mode("overwrite")

.option("delimiter", ";")

.option("header", True)

.option("compression", "gzip")

.csv("output/people_custom")

```

Or writing Parquet with partitioning (see next section) or custom schema for writing. You can also chain .format("csv").save(path) instead of .csv(path) if preferred.

Handling Multi-line JSON Files

By default Spark’s JSON reader expects one JSON record per line. For a multi-line JSON file (for example, a single JSON array or object spread across lines), you must enable the multiLine option. This tells Spark to parse the entire file as one JSON object/array.

Example: Suppose data.json is a pretty-printed JSON array spanning multiple lines:

[

{"name": "Alice", "age": 30},

{"name": "Bob", "age": 35}

]

Read it with:

```

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ExportPeople").getOrCreate()

df = spark.read.schema(schema).json("s3a://pyspark420/pyspark-upload/data.json")

df.show()

```

This correctly parses the array and loads each element as a row in the DataFrame.

Spark’s JSON reader automatically infers the schema. Remember that without multiLine=True, Spark would try to parse each line separately and likely fail on a multi-line JSON. The multiLine option is only needed for JSON; for CSV, Spark also has a multiLine option to allow line breaks inside quoted fields, but that’s less commonly used.

Partitioning and Bucketing When Writing Data

Partitioning and bucketing are ways to organize output data for efficiency.

Partitioning: Using df.write.partitionBy(col1, col2, …) saves data into a directory structure based on column values. For example:

```

hide

df.write.mode("overwrite").partitionBy("country", "gender").parquet("output/people_part")

```

This creates folders like country=US/gender=male/part-*.parquet, etc. Spark will automatically include the partition columns (country, gender) in the schema when reading. Partitioning helps queries that filter on those columns to skip reading irrelevant partitions.

Example:

```

hide

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ExportPeople").getOrCreate()

df = spark.createDataFrame([

("Alice", 30, "US", "F"),

("Bob", 35, "US", "M"),

("Carol", 40, "UK", "F")

], ["name","age","country","gender"])

df.write.mode("overwrite").partitionBy("country", "gender").parquet("output/people_part")

```

The above will create a directory tree like:

output/people_part/

├── country=US/

│ ├── gender=F/ (contains Alice)

│ └── gender=M/ (contains Bob)

└── country=UK/

└── gender=F/ (contains Carol)

When you later read with spark.read.parquet("output/people_part"), Spark will automatically load country and gender as columns.

Bucketing: Bucketing groups data into a fixed number of “buckets” by hashing one or more columns. In Spark, bucketing is used with Hive-style tables via saveAsTable. Bucketing is not used when writing plain files (it only applies with saveAsTable). For example:

```

hide

spark.sql("DROP TABLE IF EXISTS bucketed_table")

(spark.createDataFrame([(100,"Alice"),(120,"Alice"),(140,"Bob")], ["age","name"])

.write.bucketBy(2, "name")

.mode("overwrite")

.saveAsTable("bucketed_table"))

spark.read.table("bucketed_table").show()

```

This creates a Hive table “bucketed_table” with 2 buckets on the name column. Bucketing ensures that all rows with the same name end up in the same bucket-file (useful for efficient joins). Note that Spark’s bucketing uses a different hash than Hive’s, and it only works with table write (not plain file writes).

Compression Formats

Spark allows compressing output files in formats like CSV, JSON, and Parquet. You can specify a compression codec using. option("compression", "codec") or by passing the compression parameter to .csv(), .json(), or .parquet() (many APIs accept a compression argument).

Common codecs: Spark supports codecs like gzip, snappy, lzo, brotli, lz4, zstd, etc. For CSV and JSON outputs, use short names like "gzip" or "snappy". For Parquet, the default is usually Snappy (snappy), but you can override it by .option("compression","gzip") or by df.write.parquet(path, compression='gzip'). The full list of Parquet codecs Spark accepts includes: none, uncompressed, snappy, gzip, lzo, brotli, lz4, lz4_raw, and zstd.
Example (CSV with GZIP):
df.write.option("compression", "gzip").csv("output/people_gzip_csv", header=True)

This will write CSV files compressed with GZIP.

Example (Parquet with Snappy):
df.write.parquet("output/people_snappy", mode="overwrite", compression="snappy")

This writes Parquet using Snappy (default) compression.

When to use which: Gzip typically achieves higher compression ratios but is slower; Snappy is faster with reasonable compression. For very large data, Snappy is often a good default. Choose the codec based on your performance and storage needs.

Reading CSV, JSON, and Parquet Files