Spark: DataFrame

​

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 RDDs, DataFrames have a defined schema. This means that each column has a specific data type (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, 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: Transformations on DataFrames are lazy, meaning they are not executed until an action is called. This allows Spark to combine multiple transformations into a single optimized execution plan.

​

Creating DataFrames

• 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

# 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

​

QnA

df.createOrReplaceTempView("employee")  
spark.sql("SELECT * FROM employee WHERE Age > 30").show()