Apply groupby aggregate to original table using databricks api (SQL or pyspark)

FITFLOP

3 min read

·

01-10-2024

2

0

Share

Applying Groupby Aggregations to Your Databricks Tables: A Comprehensive Guide

Working with large datasets in Databricks often involves summarizing data using groupby operations and aggregations. This guide provides a comprehensive explanation of how to efficiently apply groupby aggregations to your tables using the Databricks API, both with SQL and PySpark.

Let's dive into an example scenario to illustrate the process.

Scenario: Imagine you have a table named "sales_data" containing information about sales transactions. The table has columns like "product_id", "quantity_sold", "sales_date", and "region". You want to find the total quantity of each product sold in each region.

Original Code (SQL):

SELECT product_id, region, SUM(quantity_sold) AS total_quantity_sold
FROM sales_data
GROUP BY product_id, region
ORDER BY product_id, region;

Understanding the Code

• SELECT product_id, region, SUM(quantity_sold) AS total_quantity_sold: This line specifies the columns we want to select in the final output. We're selecting the product_id, region, and the sum of quantity_sold which we've renamed as total_quantity_sold.
• FROM sales_data: This indicates the table we're pulling data from.
• GROUP BY product_id, region: This crucial part instructs the query to group rows based on the combination of product_id and region. This means all rows with the same product_id and region will be grouped together.
• ORDER BY product_id, region: This line specifies the order in which the resulting rows should be displayed. In this case, we're sorting by product_id and then by region.

Running the Code with Databricks API (SQL)

1. Connect to your Databricks cluster using the appropriate libraries like databricks-connect or spark-sql.
2. Create a SparkSession: If you're using the databricks-connect library, you can create a SparkSession as follows:

   from databricks import sql
   from databricks.sql.utils import *
   
   # Replace with your cluster ID
   cluster_id = 'your-cluster-id'
   spark = sql.connect(host='your-host',
                      http_path='your-http-path',
                      token='your-token',
                      cluster_id=cluster_id)
   

3. Execute the SQL Query:

   spark.sql("""SELECT product_id, region, SUM(quantity_sold) AS total_quantity_sold
              FROM sales_data
              GROUP BY product_id, region
              ORDER BY product_id, region""")
   

4. Access the results: The query results can be accessed as a Pandas DataFrame using the .toPandas() method. You can then further process or visualize the results as needed.

Using PySpark for Groupby Aggregations

PySpark offers a more object-oriented approach to handling data. Here's how you can achieve the same result as the SQL query:

from pyspark.sql import SparkSession

# Create a SparkSession
spark = SparkSession.builder.appName("groupby_agg").getOrCreate()

# Read the data from the table
sales_df = spark.read.table("sales_data")

# Apply groupby aggregation
result_df = sales_df.groupBy("product_id", "region").agg({"quantity_sold":"sum"}).orderBy("product_id", "region")

# Show the results
result_df.show()

Understanding the PySpark Code

• sales_df = spark.read.table("sales_data"): This line reads the "sales_data" table into a PySpark DataFrame called sales_df.
• result_df = sales_df.groupBy("product_id", "region").agg("quantity_sold""sum"): This performs the groupby operation. groupBy("product_id", "region") groups the data based on the specified columns. agg({"quantity_sold":"sum"}) applies the sum aggregation function to the quantity_sold column.
• result_df.orderBy("product_id", "region"): This line orders the resulting DataFrame by product_id and then by region.
• result_df.show(): This displays the first 20 rows of the result DataFrame in the console.

Practical Examples and Considerations

• Multiple Aggregations: You can apply multiple aggregation functions within the agg method. For example, you could find both the sum and average of quantity_sold for each product and region:

  result_df = sales_df.groupBy("product_id", "region").agg({"quantity_sold":"sum", "quantity_sold":"avg"})
  

• Filtering Before Aggregation: If you only need to analyze specific subsets of your data, you can filter the DataFrame before applying the groupby aggregation:

  filtered_sales_df = sales_df.filter(sales_df.sales_date >= "2023-01-01")
  result_df = filtered_sales_df.groupBy("product_id", "region").agg({"quantity_sold":"sum"})
  

Conclusion

This guide has demonstrated how to utilize the Databricks API for applying groupby aggregations to your tables, both with SQL and PySpark. This powerful functionality allows you to gain valuable insights and summaries from your datasets. Whether you're working with large datasets or smaller tables, the Databricks API provides you with the tools to efficiently analyze and interpret your data.