Post #187
2.86K
- ❤ 7
DA Showing posts older than #188 · Back to latest

Data Engineers Working with PySpark Aggregations What are Aggregations? Aggregations in PySpark allow you to transform large datasets by computing statistics across specified groups. PySpark offers built-in functions for common aggregations, such as sum, avg, min, max…


# Load the DataFrame
df = spark.read.csv("path/to/data.csv", header=True, inferSchema=True)
# Handle missing values
df_filled = df.fillna(0)
# Aggregate data
from pyspark.sql.functions import sum, col
df_aggregated = df_filled.groupBy("category", "region").agg(sum(col("sales")).alias("total_sales"))
# Sort the results
df_aggregated_sorted = df_aggregated.orderBy("total_sales", ascending=False)
# Save the aggregated DataFrame
df_aggregated_sorted.write.csv("path/to/aggregated/data.csv", header=True)
# Load the DataFrame
df = spark.read.csv("path/to/data.csv", header=True, inferSchema=True)
# Convert column to timestamp
from pyspark.sql.functions import to_timestamp, col
df_transformed = df.withColumn("date_column", to_timestamp(col("date_column"), "yyyy-MM-dd"))
# Handle invalid dates
df_transformed_filtered = df_transformed.filter(col("date_column").isNotNull())
# Extract date components
from pyspark.sql.functions import year, month, dayofmonth
df_transformed_extracted = df_transformed_filtered.withColumn("year", year(col("date_column"))).withColumn("month", month(col("date_column"))).withColumn("day", dayofmonth(col("date_column")))
# Save the transformed DataFrame
df_transformed_extracted.write.csv("path/to/transformed/data.csv", header=True)
# Load the DataFrame
df = spark.read.csv("path/to/data.csv", header=True, inferSchema=True)
# Partition by date
df_partitioned = df.repartitionByRange("date_column")
# Save to parquet format
df_partitioned.write.parquet("path/to/partitioned/data.parquet", partitionBy=["date_column"])
# Optimize storage
df_partitioned.write.option("compression", "snappy").parquet("path/to/partitioned/data.parquet", partitionBy=["date_column"])
