How to Compute Summary Statistics for a PySpark DataFrame: The Ultimate Guide
Published on April 17, 2025
Diving Straight into Computing Summary Statistics for a PySpark DataFrame
Need to compute summary statistics—like mean, min, max, or standard deviation—for a PySpark DataFrame to understand data distributions or validate an ETL pipeline? Calculating summary statistics is a fundamental skill for data engineers and analysts working with Apache Spark. These metrics provide quick insights into numerical and categorical data, enabling data quality checks, exploratory analysis, and informed decision-making. This guide offers an in-depth exploration of the syntax and steps for computing summary statistics in a PySpark DataFrame, with detailed examples covering basic, column-specific, grouped, nested, and SQL-based scenarios. We’ll address key errors and performance considerations to keep your pipelines robust, targeting a comprehensive ~2,000-word explanation. Let’s uncover those stats! For a foundational understanding of PySpark, see Introduction to PySpark.
Understanding Summary Statistics in PySpark
Summary statistics summarize a dataset’s key characteristics, such as central tendency (mean, median), dispersion (standard deviation, variance), and range (min, max). In PySpark, these statistics are computed across DataFrame columns, leveraging Spark’s distributed computing to handle large-scale data efficiently. Common statistics include:
- Count: Number of non-null values
- Mean: Average value
- Standard Deviation: Measure of data spread
- Min/Max: Smallest and largest values
- Quantiles: Values at specific percentiles (e.g., 25th, 50th, 75th)
PySpark provides multiple methods to compute these, including describe(), summary(), aggregation functions (e.g., mean(), stddev()), and SQL queries. These methods are essential for tasks like data profiling, outlier detection, and preparing data for machine learning. The distributed nature of Spark ensures scalability, but careful optimization is needed to manage computational costs.
Computing Basic Summary Statistics for a DataFrame
The simplest way to compute summary statistics for a PySpark DataFrame is the describe() method, which generates a new DataFrame with statistics (count, mean, stddev, min, max) for all numeric columns and count for non-numeric columns. The SparkSession, Spark’s unified entry point, orchestrates these computations across distributed data, making it ideal for quick data profiling in ETL pipelines. Here’s the basic syntax:
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("SummaryStatistics").getOrCreate()
df = spark.createDataFrame(data, schema)
stats_df = df.describe()Let’s apply it to an employee DataFrame with IDs, names, ages, salaries, and departments to compute basic statistics:
from pyspark.sql import SparkSession
# Initialize SparkSession
spark = SparkSession.builder.appName("SummaryStatistics").getOrCreate()
# Create DataFrame
data = [
("E001", "Alice", 25, 75000.0, "HR"),
("E002", "Bob", 30, 82000.5, "IT"),
("E003", "Cathy", 28, 90000.75, "HR"),
("E004", "David", 35, 100000.25, "IT"),
("E005", "Eve", 28, 78000.0, "Finance")
]
df = spark.createDataFrame(data, ["employee_id", "name", "age", "salary", "department"])
# Compute basic statistics
stats_df = df.describe()
stats_df.show(truncate=False)Output (formatted for readability):
+-------+-----------+-----+------------------+------------------+----------+
|summary|employee_id|name |age |salary |department|
+-------+-----------+-----+------------------+------------------+----------+
|count |5 |5 |5 |5 |5 |
|mean |null |null |29.2 |85000.3 |null |
|stddev |null |null |3.8340579025364633|10175.735297748765|null |
|min |E001 |Alice|25 |75000.0 |Finance |
|max |E005 |Eve |35 |100000.25 |IT |
+-------+-----------+-----+------------------+------------------+----------+This output shows:
- Count: 5 non-null values for all columns.
- Mean: Average age (29.2) and salary (85,000.3).
- Stddev: Standard deviation for age (~3.83) and salary (~10,175.74).
- Min/Max: Range for age (25–35), salary (75,000–100,000.25), and lexicographical range for strings (employee_id, name, department).
The describe() method automatically handles numeric columns (age, salary) fully and provides only count, min, and max for non-numeric columns (employee_id, name, department). To validate, check the count and ensure numerical accuracy:
assert stats_df.filter(col("summary") == "count").select("age").collect()[0]["age"] == "5", "Incorrect count"
assert abs(float(stats_df.filter(col("summary") == "mean").select("salary").collect()[0]["salary"]) - 85000.3) < 0.1, "Incorrect mean salary"This confirms the row count and mean salary. The describe() method is efficient for small to medium datasets but may require optimization for very large DataFrames due to the full data scan it performs.
Error to Watch: Empty DataFrame fails to produce meaningful statistics:
try:
empty_df = spark.createDataFrame([], schema=["employee_id", "name", "age", "salary"])
empty_df.describe().show()
except Exception as e:
print(f"Error: {e}")Output (no error, but empty result):
+-------+-----------+----+---+------+
|summary|employee_id|name|age|salary|
+-------+-----------+----+---+------+
+-------+-----------+----+---+------+Fix: Ensure the DataFrame has data:
assert df.count() > 0, "DataFrame is empty"This check prevents meaningless results from empty DataFrames.
Computing Custom Summary Statistics for Specific Columns
The describe() method provides a fixed set of statistics, but you may need custom metrics (e.g., median, specific quantiles) or statistics for specific columns. The agg() method with aggregation functions like mean(), stddev(), min(), max(), or approxQuantile() allows tailored computations. This extends basic statistics for targeted ETL analysis, such as profiling key numerical columns, as discussed in DataFrame Operations.
Let’s compute custom statistics (mean, standard deviation, and 50th percentile) for the salary column:
from pyspark.sql import SparkSession
from pyspark.sql.functions import mean, stddev, col
spark = SparkSession.builder.appName("CustomStats").getOrCreate()
# Create DataFrame
data = [
("E001", "Alice", 25, 75000.0, "HR"),
("E002", "Bob", 30, 82000.5, "IT"),
("E003", "Cathy", 28, 90000.75, "HR"),
("E004", "David", 35, 100000.25, "IT"),
("E005", "Eve", 28, 78000.0, "Finance")
]
df = spark.createDataFrame(data, ["employee_id", "name", "age", "salary", "department"])
# Compute custom statistics for salary
stats_df = df.agg(
mean("salary").alias("mean_salary"),
stddev("salary").alias("stddev_salary"),
approxQuantile("salary", [0.5], 0.01)[0].alias("median_salary")
)
stats_df.show(truncate=False)Output (approximate, as approxQuantile may vary slightly):
+-----------+--------------+-------------+
|mean_salary|stddev_salary |median_salary|
+-----------+--------------+-------------+
|85000.3 |10175.73529775|82000.5 |
+-----------+--------------+-------------+This computes the mean (~85,000.3), standard deviation (~10,175.74), and approximate median (~82,000.5) for salary. The approxQuantile() function estimates quantiles (here, the 50th percentile) with a specified relative error (0.01), balancing accuracy and performance. Validate the results:
assert abs(stats_df.collect()[0]["mean_salary"] - 85000.3) < 0.1, "Incorrect mean salary"
assert stats_df.collect()[0]["median_salary"] in [82000.5, 90000.75], "Median out of expected range"This ensures the mean is accurate and the median is reasonable (it may vary slightly due to approximation). Custom aggregations are powerful for focusing on specific columns or metrics, avoiding the overhead of computing statistics for all columns as describe() does.
Error to Watch: Aggregating non-numeric columns with numeric functions fails:
try:
stats_df = df.agg(mean("name").alias("mean_name"))
stats_df.show()
except Exception as e:
print(f"Error: {e}")Output:
Error: 'mean' is not defined for string typeFix: Verify column type:
assert df.schema["salary"].dataType.typeName() in ["double", "float", "integer", "long"], "Invalid type for mean"This ensures salary is numeric, suitable for mean() and similar functions.
Computing Summary Statistics by Group
Grouping a DataFrame by a column, like department, and computing summary statistics per group is essential for segmented analysis in ETL pipelines, such as comparing salary distributions across departments, as discussed in DataFrame Operations. Combine groupBy() with agg() to apply statistics to each group:
from pyspark.sql import SparkSession
from pyspark.sql.functions import mean, stddev, min, max
spark = SparkSession.builder.appName("GroupedStats").getOrCreate()
# Create DataFrame
data = [
("E001", "Alice", 25, 75000.0, "HR"),
("E002", "Bob", 30, 82000.5, "IT"),
("E003", "Cathy", 28, 90000.75, "HR"),
("E004", "David", 35, 100000.25, "IT"),
("E005", "Eve", 28, 78000.0, "Finance")
]
df = spark.createDataFrame(data, ["employee_id", "name", "age", "salary", "department"])
# Group by department and compute statistics
stats_df = df.groupBy("department").agg(
mean("salary").alias("mean_salary"),
stddev("salary").alias("stddev_salary"),
min("salary").alias("min_salary"),
max("salary").alias("max_salary")
)
stats_df.show(truncate=False)Output:
+----------+-----------+--------------+----------+----------+
|department|mean_salary|stddev_salary |min_salary|max_salary|
+----------+-----------+--------------+----------+----------+
|HR |82500.375 |10606.61406276|75000.0 |90000.75 |
|IT |91000.375 |12727.92206136|82000.5 |100000.25 |
|Finance |78000.0 |null |78000.0 |78000.0 |
+----------+-----------+--------------+----------+----------+This groups by department and computes the mean, standard deviation, minimum, and maximum salaries for each department. For example, the HR department has an average salary of ~82,500.375, with a standard deviation of ~10,606.61, indicating moderate variability. The Finance department’s standard deviation is null because it has only one record, insufficient for variance calculation.
Validate the results:
it_row = stats_df.filter(col("department") == "IT").collect()[0]
assert abs(it_row["mean_salary"] - 91000.375) < 0.1, "IT mean salary incorrect"
assert stats_df.count() == 3, "Unexpected department count"This confirms the IT department’s mean salary and the number of departments. Grouping with statistics is computationally intensive due to shuffling, so optimize by filtering irrelevant rows or partitioning by the grouping column (repartition("department")) for large datasets.
Error to Watch: Grouping by a non-existent column fails:
try:
stats_df = df.groupBy("invalid_column").agg(mean("salary").alias("mean_salary"))
stats_df.show()
except Exception as e:
print(f"Error: {e}")Output:
Error: Column 'invalid_column' does not existFix: Verify grouping column:
assert "department" in df.columns, "Grouping column missing"This ensures the department column exists before grouping.
Computing Statistics for Nested Data
Nested DataFrames, with structs or arrays, model complex relationships, such as employee contact details or project lists. Computing statistics on nested fields, like counting non-null emails or aggregating project counts, extends grouped statistics for advanced ETL analytics, as discussed in DataFrame UDFs. Access nested fields using dot notation (e.g., contact.email):
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, LongType, ArrayType
from pyspark.sql.functions import count, sum as sum_agg
spark = SparkSession.builder.appName("NestedStats").getOrCreate()
# Define schema with nested structs
schema = StructType([
StructField("employee_id", StringType(), False),
StructField("name", StringType(), True),
StructField("contact", StructType([
StructField("phone", LongType(), True),
StructField("email", StringType(), True)
]), True),
StructField("department", StringType(), True)
])
# Create DataFrame
data = [
("E001", "Alice", (1234567890, "alice@example.com"), "HR"),
("E002", "Bob", (9876543210, None), "IT"),
("E003", "Cathy", (None, "cathy@example.com"), "HR"),
("E004", "David", (5555555555, "david@example.com"), "IT")
]
df = spark.createDataFrame(data, schema)
# Compute statistics for nested field
stats_df = df.groupBy("department").agg(
count("contact.email").alias("email_count"),
count("contact.phone").alias("phone_count")
)
stats_df.show(truncate=False)Output:
+----------+-----------+-----------+
|department|email_count|phone_count|
+----------+-----------+-----------+
|HR |2 |1 |
|IT |1 |2 |
+----------+-----------+-----------+This groups by department and counts non-null contact.email and contact.phone values per department. For example, HR has two non-null emails (Alice, Cathy) but only one non-null phone (Alice), reflecting the data’s null patterns. Validate:
hr_row = stats_df.filter(col("department") == "HR").collect()[0]
assert hr_row["email_count"] == 2 and hr_row["phone_count"] == 1, "HR nested stats incorrect"This confirms the HR department’s counts. When working with nested data, ensure null handling aligns with your analysis goals, as count() excludes nulls by default. For more complex nested aggregations, consider flattening structs or exploding arrays before grouping.
Error to Watch: Invalid nested field fails:
try:
stats_df = df.groupBy("department").agg(count("contact.invalid_field").alias("count"))
stats_df.show()
except Exception as e:
print(f"Error: {e}")Output:
Error: StructField 'contact' does not contain field 'invalid_field'Fix: Validate nested field:
assert "email" in [f.name for f in df.schema["contact"].dataType.fields], "Nested field missing"This ensures the contact.email field exists before aggregation.
Computing Statistics Using SQL Queries
For teams familiar with SQL or pipelines integrating with SQL-based tools, using a SQL query via a temporary view to compute statistics offers a powerful alternative, extending nested statistics for SQL-driven ETL workflows, as seen in DataFrame Operations. Temporary views make DataFrames queryable like database tables, leveraging SQL’s expressive syntax:
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("SQLStats").getOrCreate()
# Create DataFrame
data = [
("E001", "Alice", 25, 75000.0, "HR"),
("E002", "Bob", 30, 82000.5, "IT"),
("E003", "Cathy", 28, 90000.75, "HR"),
("E004", "David", 35, 100000.25, "IT"),
("E005", "Eve", 28, 78000.0, "Finance")
]
df = spark.createDataFrame(data, ["employee_id", "name", "age", "salary", "department"])
# Create temporary view
df.createOrReplaceTempView("employees")
# Compute statistics using SQL
stats_df = spark.sql("""
SELECT department,
COUNT(employee_id) AS employee_count,
AVG(salary) AS mean_salary,
STDDEV(salary) AS stddev_salary,
MIN(salary) AS min_salary,
MAX(salary) AS max_salary
FROM employees
GROUP BY department
""")
stats_df.show(truncate=False)Output:
+----------+--------------+-----------+--------------+----------+----------+
|department|employee_count|mean_salary|stddev_salary |min_salary|max_salary|
+----------+--------------+-----------+--------------+----------+----------+
|HR |2 |82500.375 |10606.61406276|75000.0 |90000.75 |
|IT |2 |91000.375 |12727.92206136|82000.5 |100000.25 |
|Finance |1 |78000.0 |null |78000.0 |78000.0 |
+----------+--------------+-----------+--------------+----------+----------+This SQL query groups by department and computes the employee count, mean salary, standard deviation, minimum, and maximum salaries per department. The syntax mirrors standard SQL, making it accessible for database-savvy teams. Validate:
it_row = stats_df.filter(col("department") == "IT").collect()[0]
assert abs(it_row["mean_salary"] - 91000.375) < 0.1 and it_row["employee_count"] == 2, "IT stats incorrect"
assert stats_df.count() == 3, "Unexpected department count"This confirms the IT department’s mean salary and employee count, as well as the number of departments. SQL queries are particularly useful when integrating Spark with existing SQL-based workflows or when stakeholders prefer SQL’s readability.
Error to Watch: Querying an unregistered view fails:
try:
stats_df = spark.sql("SELECT department, COUNT(*) FROM nonexistent GROUP BY department")
stats_df.show()
except Exception as e:
print(f"Error: {e}")Output:
Error: Table or view not found: nonexistentFix: Ensure the view is registered:
assert "employees" in [v.name for v in spark.catalog.listTables()], "View missing"
df.createOrReplaceTempView("employees")This prevents errors by verifying the employees view exists.
Performance Considerations for Computing Statistics
Computing summary statistics in PySpark involves scanning the entire dataset, which can be costly for large DataFrames. Additionally, grouping operations introduce data shuffling, further impacting performance. To optimize, consider these best practices:
- Filter Early: Apply filters to reduce the dataset before computing statistics. For example, exclude invalid or outlier rows:
df = df.filter(col("salary") > 0) # Exclude invalid salaries- Select Relevant Columns: Use select() to include only columns needed for statistics, minimizing data processed:
df = df.select("salary", "department")- Partition Data: Repartition by the grouping column to reduce shuffling:
df = df.repartition("department")- Cache Results: If statistics are reused, cache the DataFrame:
stats_df.cache()- Use Approximate Functions: For quantiles, approxQuantile() is faster than exact methods, trading precision for performance:
median = df.approxQuantile("salary", [0.5], 0.01)[0]For example, to optimize grouped statistics, filter and repartition:
optimized_df = df.filter(col("salary") > 76000).repartition("department")
stats_df = optimized_df.groupBy("department").agg(
mean("salary").alias("mean_salary"),
stddev("salary").alias("stddev_salary")
)
stats_df.show(truncate=False)This reduces the dataset and aligns partitions with department, minimizing shuffle overhead. Monitor performance via the Spark UI, checking shuffle read/write metrics to identify bottlenecks.
Practical Applications of Summary Statistics
Summary statistics are indispensable for various ETL and analytics tasks:
- Data Profiling: Use describe() to inspect data distributions, identifying outliers or missing values before processing.
- Quality Assurance: Validate data integrity by checking counts and ranges, ensuring no unexpected nulls or anomalies.
- Exploratory Analysis: Compute means, medians, and standard deviations to understand trends, such as salary variability across departments.
- Feature Engineering: Generate aggregated features (e.g., average purchase per customer) for machine learning models.
- Reporting: Produce department-level summaries for dashboards, combining counts, sums, and averages for stakeholder insights.
These applications underscore the versatility of summary statistics, making them a cornerstone of data engineering workflows.
Advanced Statistics with Custom UDFs
For specialized statistics not covered by built-in functions, User-Defined Functions (UDFs) allow custom logic, extending the flexibility of summary statistics. For example, suppose you want to compute a custom weighted average salary per department, factoring in a hypothetical seniority weight based on age. You can create a UDF to implement this:
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, col, sum as sum_agg
from pyspark.sql.types import DoubleType
spark = SparkSession.builder.appName("UDFStats").getOrCreate()
# Create DataFrame
data = [
("E001", "Alice", 25, 75000.0, "HR"),
("E002", "Bob", 30, 82000.5, "IT"),
("E003", "Cathy", 28, 90000.75, "HR"),
("E004", "David", 35, 100000.25, "IT")
]
df = spark.createDataFrame(data, ["employee_id", "name", "age", "salary", "department"])
# Define UDF for weighted average (weight = age)
def weighted_salary(salary, age):
return salary * (age / 30.0) # Normalize age relative to 30
weighted_udf = udf(weighted_salary, DoubleType())
# Add weighted salary column
df = df.withColumn("weighted_salary", weighted_udf(col("salary"), col("age")))
# Group and aggregate with weighted salary
stats_df = df.groupBy("department").agg(
sum_agg("weighted_salary").alias("total_weighted_salary"),
sum_agg("salary").alias("total_salary")
)
stats_df.show(truncate=False)Output (approximate, as weighted values depend on age):
+----------+--------------------+------------+
|department|total_weighted_salary|total_salary|
+----------+--------------------+------------+
|HR |140833.9375 |165000.75 |
|IT |168333.58333333334 |182000.75 |
+----------+--------------------+------------+This computes a weighted salary (salary scaled by age/30) and sums it per department, alongside the regular total salary. The UDF multiplies each salary by a weight based on age, simulating seniority. Validate:
hr_row = stats_df.filter(col("department") == "HR").collect()[0]
assert abs(hr_row["total_salary"] - 165000.75) < 0.1, "HR total salary incorrect"UDFs are flexible but slower due to serialization overhead. Use native functions like sum() or avg() when possible, reserving UDFs for unique calculations. For nested data, UDFs can also process structs or arrays, but flattening or exploding may simplify aggregation.
How to Fix Common Summary Statistics Errors
Errors can disrupt statistic computations. Here are key issues, with fixes:
- Non-Existent Column: Aggregating invalid columns fails. Fix:
assert column in df.columns, "Column missing"- Invalid Nested Field: Aggregating invalid nested fields fails. Fix:
assert field in [f.name for f in df.schema[nested_col].dataType.fields], "Nested field missing"- Non-Existent View: SQL on unregistered views fails. Fix:
assert view_name in [v.name for v in spark.catalog.listTables()], "View missing"
df.createOrReplaceTempView(view_name)- Incompatible Aggregation Type: Using numeric functions on strings fails. Fix:
assert df.schema[column].dataType.typeName() in ["double", "float", "integer", "long"], "Invalid type"- Empty DataFrame: Computing statistics on empty DataFrames produces empty or null results. Fix:
assert df.count() > 0, "DataFrame empty"- Performance Issues: Full scans or shuffling slow down large datasets. Fix: Filter and partition:
df = df.filter(col("salary") > 0).repartition("department")These checks ensure robust and efficient statistic computations, minimizing errors and optimizing performance.
Wrapping Up Your Summary Statistics Mastery
Computing summary statistics for a PySpark DataFrame is a transformative skill that unlocks data profiling, quality assurance, and analytical insights. Whether you’re using describe() for quick overviews, agg() for custom metrics, groupBy() for segmented analysis, nested field aggregations, custom UDFs for specialized calculations, or SQL queries for familiar syntax, Spark provides versatile tools to handle diverse ETL scenarios. By mastering these techniques, optimizing performance, and anticipating errors, you can build efficient, reliable pipelines that turn raw data into actionable knowledge. These methods will elevate your data engineering workflows, enabling you to tackle complex analytics with confidence.
Try these approaches in your next Spark job, and share your experiences, tips, or questions in the comments or on X. Keep exploring with DataFrame Operations to deepen your PySpark expertise!