CSVs are ubiquitous. You'll find them everywhere.

If you're using Jupyter notebook setup, here is the code I used to initialize things:

# for jupyter notebook
from IPython.core.interactiveshell import InteractiveShell
InteractiveShell.ast_node_interactivity='all'

# import relavant python modules
from pyspark.sql import types as T
from pyspark.sql import functions as F

# Create spark session (not required on databricks)
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local[*]").getOrCreate()

Read CSV

The barebones read method template is:

sales_df = (
    spark
    .read
    .format("csv") # set data type format
    .load("/data/sales/") # set load location   
)

i

Note: Instead of using format().load() to set the data format type and data location, you can directly use the format-specific csv() to set both data format and location through a single method.

sales_df = (
    spark
    .read
    .csv("/data/sales/")    
)

However, it is recommended that you stick to using the standard template of .format().load(). It keeps your codebase more uniform.

In this article, I'm sticking to the .format().load() way of loading csv.

A more practical read command will:

• Set column names from the csv files' header
• Automatically infer schema from loaded data (schema on read)

sales_df = (
    spark
    .read
    .format("csv")
    .option("header", "true")
    .option("inferschema", "true") # optional option
    .load("/data/sales/")
)

Set schema

Infer schema is not always accurate. Sometimes, you simply know your data better than the spark engine. In this case, set the dataframe's schema manually.

Keep in mind that you're defining the schema of the dataframe (the data that has been loaded into the dataframe). You aren't defining the schema of the source file.

Automatically set schema using inferSchema

sales_df = (
    spark
    .read
    .format("csv")
    .option("header", "true")
    .option("inferschema", "true") # optional option
    .load("/data/sales/")
)

Manually set schema

There are 2 ways to set schema manually:

1. Using DDL string
2. Programmatically, using StructType and StructField

Set schema using DDL string

This is the recommended way to define schema, as it is the easier and more readable option. These datatypes we use in the string are the Spark SQL datatypes.

The format is simple. It is a string-csv of the dataframe's every column name & datatype.

schema_ddl_string = "<column_name> <data type>, <column_name> <data type>, <column_name> <data type>"

sales_schema_ddl = """year INT, month INT, day INT, amount DECIMAL"""

sales_df = (
    spark
    .read
    .format("csv")
    .schema(sales_schema_ddl) # set schema
    .option("header", "true")
    .load("/data/sales/")
)

The datatypes you can use for DDL string are:

Source - Spark SQL - Supported Data Types

Set schema using python API functions

We define a column's datatype using the StructField() method.

To define the schema of the entire dataframe, we -

• Create a StructField("column_name", DataType()) that defines datatype of one column of the dataframe
• Create a list of StructField(), to define the data type of every column in the dataframe
• Pass list of StructFields into a StructType()
• Pass the StructType() object to the schema() method of DataFrameReader.read

It looks easier if you directly look at the code:

from pyspark.sql import types as T

sales_schema_struct = T.StructType(
    [
        T.StructField("year", T.IntegerType()),
        T.StructField("month", T.IntegerType()),
        T.StructField("day", T.IntegerType()),
        T.StructField("amount", T.DecimalType()),
    ]
)

sales_df = (
    spark
    .read.format("csv")
    .schema(sales_schema_struct)  # set schema
    .option("header", "true")
    .load("/data/sales/")
)

The datatypes you can use in StructField() are:

Source - Spark SQL - Supported Data Types

Write CSV

The barebones write method template is:

sales_df.write.format("csv").save("/data/sales")

A more practical write command will:

• Save column names into the file's header
• Explicitly set write mode

(
    sales_df
    .write
    .format("csv")
    .option("header", "true")
    .mode("overwrite")
    .save("/data/sales")
)

Partition written data on specific columns

To make reading data faster and more optimized, by increasing parallelization.

Eg - Partition by year, month, date:

(
    sales_df
    .write
    .format("csv")
    .option("header", "true")
    .mode("overwrite")
    .partitionBy("year", "date", "day")
    .save("/data/sales")
)

Bucket written data on specific columns

To break down the file sizes within each partition.

Note : save() doesn't support saving after using bucketBy(). Only saveAsTable() suports it.

If you use save(), it will throw an exception AnalysisException: 'save' does not support bucketBy right now.

Eg - Create 10 buckets within each partition:

(
    sales_df
    .write
    .format("csv")
    .mode("overwrite")
    .option("header", "true")
    .option("path", "data/sales")
    .bucketBy(10, "year", "date", "day")
    .saveAsTable("sales")
)

More details in bucketBy() documentation

List of options

Source - CSV data source options documentation

Warning on column order in csv files

WARNING FOR CSV FILES

Spark assumes that the column order is same across all csv files. So, column order should not change across csv files.

Spark doesn't intelligently check column name in source file and then insert them in the proper column in the table. It will stupidly insert nth column in source file into nth column in table. It doesn't check if the column orders in the source file are different from column orders in the destination dataframe/table.

Example:

Consider 2 csv files with different column orders:

data1.csv

age,name
20,alice
25,bob

data2.csv

name,age
chad,30
dave,35

Now, when you load the two csv files, the data looks like this:

person_df = (
    spark
    .read
    .format("csv") # set data type format
    .option("header", "true")
    .load(["/home/jovyan/data/data1.csv", "/home/jovyan/data/data2.csv"]) # set load location   
)
person_df.show()

Output:

+----+-----+
| age| name|
+----+-----+
|  20|alice|
|  25|  bob|
|chad|   30|
|dave|   35|
+----+-----+

You can see how age and name data got mixed up. So, always make sure the column order is same across all the csv files you are loading.