> ## Documentation Index
> Fetch the complete documentation index at: https://rajanand.org/llms.txt
> Use this file to discover all available pages before exploring further.

# DataFrameReader and DataFrameWriter

## Objectives

1. Read from CSV files
2. Read from JSON files
3. Write DataFrame to files
4. Write DataFrame to tables
5. Write DataFrame to Delta tables

Methods:

* DataFrameReader: `read`, `csv`, `json`, `parquet`, `table`, `format`, `load`, `option`, `schema`
* DataFrameWriter: `write`, `csv`, `json`, `parquet`, `table`, `format`, `save`, `option`, `mode`, `saveAsTable`,
* StructType: toDDL

SparkTypes:

* Types: `StructType`, `StructField`, `StringType`, `IntegerType`, `DoubleType`, `DateType`, `TimestampType`, `BooleanType`

## DataFrameReader

* The `DataFrameReader` class is used to read data from various sources (e.g., CSV, JSON, Parquet, Delta) into a DataFrame.
* Interface used to load a DataFrame from external data source.
* The `DataFrameReader` class is available as the `read` attribute of the `SparkSession` object.

`spark.read.parquet("/mnt/tables/sales")`

* The above command reads the Parquet files from the specified path and returns a DataFrame.
* The `read` method returns a `DataFrameReader` object that can be used to read data from various sources.

### Read from CSV files

* Read from CSV with the DataFrameReader class' `csv` method.

```python theme={"system"}
file_path = "/mnt/tables/sales.csv"
sales_df = spark
    .read \
    .option("sep", "\t") \
    .option("header", "true") \ 
    .option("inferSchema", "true") \
    .csv(file_path)

sales_df.printSchema()

```

When you execute the above code, you will see two jobs spun. You may wonder why when there is no action involved, the jobs spun.
The below two option that we used in the above code are required the file to be read. This is the reason you see the jobs executed.

```
    .option("header", "true") \ 
    .option("inferSchema", "true") \
```

Another way to write the same code is as below.

```python theme={"system"}
file_path = "/mnt/tables/sales.csv"
sales_df = spark.read.csv(file_path, sep="\t", header=True, inferSchema=True)
sales_df.printSchema()
```

Define the schema explicitly using `StructType` and `StructField` classes.

```python theme={"system"}
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType, DateType
from datetime import datetime
sales_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("price", DoubleType(), True),
    StructField("order_date", DateType(), True)
])
sales_df = spark.read.csv(file_path, sep="\t", header=True, schema=sales_schema)
```

* The above code reads the CSV file from the specified path and returns a DataFrame with the specified schema.
* The `StructType` class is used to define the schema of the DataFrame.
* The `StructField` class is used to define the fields of the schema.
* The `StringType`, `IntegerType`, `DoubleType`, and `DateType` classes are used to define the data types of the fields.
* The `True` parameter in the `StructField` constructor indicates that the field is nullable.
* This time there is no need to spin the jobs as we have defined the schema explicitly.

There is an alternative way to define schema using DDL syntax.

```
ddl_schema = """
    order_id STRING,
    customer_id STRING,
    product_id STRING,
    quantity INT,
    price DOUBLE,
    order_date DATE
"""
sales_df = spark.read.csv(file_path, sep="\t", header=True, schema=ddl_schema)
```

This code does exactly the same thing as the previous code, but uses DDL syntax to define the schema.

* The `ddl_schema` variable contains the DDL syntax for the schema.
* The `schema` parameter in the `csv` method is set to the `ddl_schema` variable.

### Read from JSON files

* Read from JSON with the DataFrameReader class' `json` method.

```python theme={"system"}
file_path = "/mnt/tables/sales.json"
sales_df = spark
    .read \
    .option("inferSchema", "true") \
    .json(file_path)
sales_df.printSchema()
```

* The above code reads the JSON file from the specified path and returns a DataFrame.
* This will spin a job as we have used the `inferSchema` option. If we explicitly define the schema, there will be no job spun.

You can use the `StructType` Scala method `toDDL` to convert the schema to DDL syntax. This will provide you the DDL-formatted string. This will be useful if you want to get the DDL formated string for ingesting CSV or JSON. But you don't want to write the StructType variant of the schema.
This functionality is not available in Python. But you can use the `toDDL` method in Scala and then copy the DDL-formatted string to Python.

```scala theme={"system"}
val sales_file_path = "/mnt/tables/sales.json"
val sales_df_schema = spark.read.option("inferSchema", "true").json(sales_file_path).schema.toDDL()
println(sales_df_schema)
```