How to read and write CSV files with Spark

CSV files are a common handoff format for exports, reports, and batch feeds that still need to pass through Apache Spark. Reading them into a DataFrame with an explicit schema and writing the filtered result back to CSV keeps separators, headers, and row counts visible during the handoff.

PySpark uses Spark SQL's built-in CSV data source through the DataFrame reader and writer. The local job uses a header row, a fixed schema, and FAILFAST parsing so field types and malformed rows are handled before the output directory is written.

Spark writes CSV output as a directory of part files plus commit metadata, not as one file path. A small order dataset is narrowed to APAC priority rows, written as one part file for inspection, and read back so the row count and values prove the write.

1. Create a CSV input file with headers and one quoted comma value.

   $ cat > orders.csv <<'CSV'
   order_id,region,item_count,order_total,notes
   ORD-1001,APAC,3,127.50,"bulk, expedited"
   ORD-1002,EMEA,1,19.99,standard
   ORD-1003,APAC,7,233.10,invoice reviewed
   CSV

   The quoted comma in bulk, expedited confirms that the parser keeps a separator inside a field instead of shifting the row.
   Tool: Comma-Separated Values (CSV) Converter

2. Create the PySpark read/write job.

   spark_csv_check.py

   from pyspark.sql import SparkSession
   from pyspark.sql import functions as F
    
    
   spark = (
       SparkSession.builder.master("local[1]")
       .appName("sg-spark-csv-read-write")
       .getOrCreate()
   )
   spark.sparkContext.setLogLevel("ERROR")
    
   schema = "order_id STRING, region STRING, item_count INT, order_total DOUBLE, notes STRING"
    
   orders = (
       spark.read.schema(schema)
       .option("header", True)
       .option("mode", "FAILFAST")
       .csv("orders.csv")
   )
    
   priority_orders = (
       orders.where((F.col("region") == "APAC") & (F.col("item_count") >= 3))
       .select("order_id", "region", "item_count", "order_total", "notes")
       .orderBy("order_id")
   )
    
   print("Input rows:", orders.count())
   orders.printSchema()
   print("Priority rows before write:")
   priority_orders.show(truncate=False)
    
   (
       priority_orders.coalesce(1)
       .write.mode("overwrite")
       .option("header", True)
       .csv("output/csv-priority-orders")
   )
    
   read_back = (
       spark.read.schema(priority_orders.schema)
       .option("header", True)
       .csv("output/csv-priority-orders")
   )
    
   print("Output rows:", read_back.count())
   read_back.orderBy("order_id").show(truncate=False)
    
   spark.stop()

   coalesce(1) keeps this tiny result to one part file for inspection. Let production jobs write multiple part files unless a downstream handoff explicitly requires a single file.

3. Run the Spark job and verify that the read-back output contains two APAC rows.

   $ spark-submit spark_csv_check.py
   ##### snipped #####
   Input rows: 3
   root
    |-- order_id: string (nullable = true)
    |-- region: string (nullable = true)
    |-- item_count: integer (nullable = true)
    |-- order_total: double (nullable = true)
    |-- notes: string (nullable = true)
   
   Priority rows before write:
   +--------+------+----------+-----------+----------------+
   |order_id|region|item_count|order_total|notes           |
   +--------+------+----------+-----------+----------------+
   |ORD-1001|APAC  |3         |127.5      |bulk, expedited |
   |ORD-1003|APAC  |7         |233.1      |invoice reviewed|
   +--------+------+----------+-----------+----------------+
   
   Output rows: 2
   +--------+------+----------+-----------+----------------+
   |order_id|region|item_count|order_total|notes           |
   +--------+------+----------+-----------+----------------+
   |ORD-1001|APAC  |3         |127.5      |bulk, expedited |
   |ORD-1003|APAC  |7         |233.1      |invoice reviewed|
   +--------+------+----------+-----------+----------------+

   The second table is read from output/csv-priority-orders after Spark writes the directory.

4. Inspect the CSV part file Spark wrote.

   $ cat output/csv-priority-orders/part-*.csv
   order_id,region,item_count,order_total,notes
   ORD-1001,APAC,3,127.5,"bulk, expedited"
   ORD-1003,APAC,7,233.1,invoice reviewed

   Spark also writes commit metadata such as _SUCCESS. Downstream tools should read the data part files or the output directory, not assume the target path is one physical file.

5. Remove the temporary files after the read-back and part-file checks pass.

   $ rm -r orders.csv spark_csv_check.py output