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

# Spark: DataFrame

A Spark DataFrame is a immutable distributed collection of data organized into named columns, similar to a table in a relational database or a data frame in R/Python pandas.
They provide a distributed, in-memory representation of tabular data, offering significant advantages over [RDD](/spark/spark-rdd)s for many data manipulation tasks.

## Key Characteristics

* **Distributed:** DataFrames are distributed across the Spark cluster, enabling parallel processing.  This allows for efficient handling of massive datasets that wouldn't fit on a single machine.

* **Schema-Aware:** Unlike <Tooltip tip="Resilient Distributed Dataset">[RDD](/spark/spark-rdd)s</Tooltip>, DataFrames have a defined schema.  This means that each column has a specific [data type](/spark/spark-datatypes) (e.g., integer, string, date).  The schema enhances data integrity and enables Spark's optimizer to generate more efficient execution plans.

* **Immutable:**  Similar to RDDs, DataFrames are immutable.  Any operation that appears to modify a DataFrame actually creates a *new* DataFrame.  This immutability is crucial for Spark's fault tolerance and simplifies reasoning about data transformations.

* **Optimized Execution:** Spark's optimizer can leverage the schema information to perform various optimizations, such as column pruning, [predicate pushdown](/spark/spark-predicate-pushdown), and join optimization.  This leads to significant performance improvements compared to RDD-based operations.

* **Integration with Spark SQL:** DataFrames seamlessly integrate with Spark SQL, allowing you to use SQL queries to process your data.  This provides a familiar and powerful way to manipulate and analyze data.

* **[Lazy Evaluation](/glossary/lazy-evaluation):** Transformations on DataFrames are lazy, meaning they are not executed until an [action](/spark/spark-transformations-vs-actions) is called.  This allows Spark to combine multiple transformations into a single optimized execution plan.

## Creating DataFrames

DataFrames can be [created](/spark/spark-create-dataframe) from various sources:

* **From existing RDDs:**  You can create a DataFrame from an RDD that contains structured data.

* **From CSV, JSON, Parquet, etc.:** Spark can directly load data from various file formats into a DataFrame.

* **From a list of dictionaries or tuples:**  You can create a DataFrame from a list of dictionaries or tuples in your driver program.  You'll typically need to specify the schema explicitly in this case.

* **From existing tables or views in a database:** Spark can connect to databases and load data from tables or views into DataFrames.

## Common DataFrame Operations

* **Selecting Columns:**  Choose specific columns using `select()`.

* **Filtering Rows:**  Filter rows based on conditions using `filter()` or `where()`.

* **Grouping and Aggregating:**  Group data using `groupBy()` and perform aggregations (e.g., `sum()`, `avg()`, `count()`, `max()`, `min()`) using `agg()`.

* **Joining DataFrames:**  Combine DataFrames based on common columns using `join()`.

* **Sorting:**  Sort data using `orderBy()`.

* **Adding/Removing Columns:**  Add new columns using `withColumn()` and remove columns using `drop()`.

* **Writing Data:**  Save the DataFrame to various formats (e.g., CSV, Parquet, JSON, database tables) using `write()`.

## Example

Let's work through a practical example using Indian census data:

```python theme={"system"}
# First, we need to import necessary libraries
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg, count, when

# Create a Spark session
spark = SparkSession.builder \
    .appName("Indian Census Analysis") \
    .getOrCreate()

# Define the schema for our DataFrame
columns = ["state", "city", "population", "area_type", "area_sqkm"]

# Let's create sample census data
census_data = [
    ("Maharashtra", "Mumbai", 20185064, "Urban", 516.7),
    ("Tamil Nadu", "Chennai", 10972000, "Urban", 426.1),
    ("Karnataka", "Bengaluru", 12327000, "Urban", 741.0),
    ("Kerala", "Malappuram", 4112920, "Rural", 441.6),
    ("Gujarat", "Ahmedabad", 8253000, "Urban", 464.2)
]

# Create DataFrame
df = spark.createDataFrame(data=census_data, schema=columns)

# Now let's perform some common operations:

# 1. Basic DataFrame operations
print("Basic DataFrame view:")
df.show()  # Display the data
df.printSchema()  # Show the structure

# 2. Select specific columns
print("\nOnly state and population:")
df.select("state", "population").show()

# 3. Filter data (e.g., find cities with population > 10 million)
print("\nCities with population > 10 million:")
df.filter(col("population") > 10000000).show()

# 4. Add a new column (population density)
print("\nAdding population density column:")
df_with_density = df.withColumn("population_density", 
                               col("population") / col("area_sqkm"))
df_with_density.show()

# 5. Group by and aggregate
print("\nAverage population by area type:")
df.groupBy("area_type") \
  .agg(avg("population").alias("avg_population")) \
  .show()

# 6. Complex conditions (categorize cities by size)
df_categorized = df.withColumn("city_category",
    when(col("population") > 15000000, "Mega City")
    .when(col("population") > 10000000, "Very Large City")
    .when(col("population") > 5000000, "Large City")
    .otherwise("Medium City"))

print("\nCities categorized by population:")
df_categorized.show()
```

## Advantages over RDDs

* **Higher-level abstraction:** DataFrames provide a more user-friendly and intuitive way to work with structured data compared to RDDs.

* **Schema enforcement:** The schema ensures data integrity and enables better optimization.

* **Improved performance:**  The optimizer can generate more efficient execution plans, leading to significant performance gains.

* **SQL integration:**  The ability to use SQL queries simplifies data manipulation.

## When to Use DataFrames

DataFrames are the preferred approach for most structured and semi-structured data processing tasks in Spark.  They offer a powerful, efficient, and user-friendly way to manipulate and analyze large datasets.  RDDs are generally used for more specialized tasks or when dealing with unstructured data where a lower-level abstraction is necessary.

## QnA

<AccordionGroup>
  <Accordion title="Q: What is a DataFrame in Spark?">
    A DataFrame in Spark is a distributed collection of data organized into named columns, similar to a table in a relational database or a data frame in R or Python (Pandas). It is an abstraction that provides a simpler API for structured data processing and integrates seamlessly with Spark SQL.\
    DataFrames are immutable and distributed, meaning they can scale across large datasets and be processed in parallel.
  </Accordion>

  <Accordion title="Q: How is a DataFrame different from an RDD?">
    A DataFrame differs from an RDD in several ways:

    * **Schema:** DataFrames have a schema, with named columns and types, whereas RDDs are just a collection of records.
    * **Optimizations:** DataFrames benefit from Catalyst Optimizer and Tungsten execution engine, making them faster than RDDs.
    * **Ease of Use:** DataFrames provide a high-level API for querying data using SQL-like syntax or Python, Scala, and Java functions. RDDs require more verbose code for the same operations.
  </Accordion>

  <Accordion title="Q: Can I use DataFrame with PySpark?">
    Yes, DataFrames are a fundamental part of PySpark. They allow you to process structured and semi-structured data efficiently using Python. PySpark DataFrames are integrated with Spark SQL, enabling you to run SQL queries and leverage DataFrame APIs for transformations and actions.
  </Accordion>

  <Accordion title="Q: How do I create a DataFrame in Spark?">
    You can create a DataFrame in multiple ways:

    1. **From an existing RDD:** Convert an RDD to a DataFrame by providing a schema.
    2. **From a file:** Read data from files like CSV, JSON, Parquet, or ORC using `spark.read` APIs.
    3. **From a database:** Load data from relational databases via JDBC.
    4. **Programmatically:** Create a DataFrame directly using Spark SQL Row objects or collections of data.
  </Accordion>

  <Accordion title="Q: Can DataFrames be used with Spark SQL?">
    Yes, DataFrames can be registered as temporary views or tables, allowing you to run SQL queries on them directly. For example:

    ```python theme={"system"}
    df.createOrReplaceTempView("employee")  
    spark.sql("SELECT * FROM employee WHERE Age > 30").show()  
    ```

    DataFrames are tightly integrated with Spark SQL, making them a preferred choice for structured data processing.
  </Accordion>

  <Accordion title="Q: Are DataFrames mutable?">
    No, DataFrames in Spark are immutable. Every transformation applied to a DataFrame creates a new DataFrame. This immutability ensures fault tolerance and makes Spark operations more predictable.
  </Accordion>

  <Accordion title="Q: How does a DataFrame improve performance?">
    DataFrames benefit from the following optimizations:

    1. **Catalyst Optimizer:** Analyzes and optimizes queries for better execution plans.
    2. **Tungsten Execution Engine:** Enhances in-memory computation and CPU utilization.
    3. **Columnar Storage:** For file formats like Parquet and ORC, Spark reads only the required columns, reducing I/O overhead.\
       These optimizations make DataFrames faster than RDDs for most operations.
  </Accordion>

  <Accordion title="Q: Can I convert between DataFrame and RDD?" defaultOpen="true">
    Yes, you can convert between DataFrame and RDD:

    * **To RDD:** Use the `.rdd` property:
      ```python theme={"system"}
      rdd = df.rdd  
      ```
    * **To DataFrame:** Use the `.toDF()` method or `createDataFrame()`:
      ```python theme={"system"}
      df = rdd.toDF(["Name", "Age"])  
      ```
  </Accordion>

  <Accordion title="Q: What file formats can DataFrames read and write?">
    DataFrames can read from and write to a variety of file formats, including:

    * CSV
    * JSON
    * Parquet
    * ORC
    * Avro
    * Delta Lake (if supported in your Spark environment)\
      You can specify the format using `spark.read.format` and `df.write.format`.
  </Accordion>
</AccordionGroup>