Fabric - Lakehouse - notebook - Reading and transforming data

Reading and transforming Lakehouse data using a notebook

We can reference spark tables in SQL cells or use dataframe methods in python cells.

In a SQL cell, we simple write SPARK SQL (a close variant of T-SQL)

%%sql

SELECT YEAR(OrderDate) AS OrderYear,

       SUM((UnitPrice * Quantity) + Tax) AS GrossRevenue

FROM salesorders

GROUP BY YEAR(OrderDate)

ORDER BY OrderYear;

To load data into a dataframe in a notebook from a spark table, open the Lakehouse, locate the table, click ...,

then select Load Data -> spark 

which will create a cell containing the following code.

df = spark.sql("SELECT * FROM lh_DBTL.salesorders LIMIT 1000")

Use tripple quotes to enable formatting of a larger SQL query (like the one used above in the SQL cell).

df = spark.sql("""

    SELECT YEAR(OrderDate) AS OrderYear,

        SUM((UnitPrice * Quantity) + Tax) AS GrossRevenue

    FROM salesorders

    GROUP BY YEAR(OrderDate)

    ORDER BY OrderYear;

""")

Passing data between a PySpark cell and a SQL cell...

%%pyspark

# Create a table in PySpark cell that can be read in a sql cell

df = spark.sql("SELECT * FROM lh_DBTL.salesorders LIMIT 1000")

df.createOrReplaceTempView("so")

%%sql

-- Read the view created in the pyspark cell in this sql cell

SELECT * from so

%%sql

-- Create a temp view in a sql cell that we will read in a pyspark cell

CREATE OR REPLACE TEMP VIEW so2 AS

    SELECT

        YEAR(OrderDate) AS OrderYear,

        SUM((UnitPrice * Quantity) + Tax) AS GrossRevenue

    FROM salesorders

    GROUP BY YEAR(OrderDate)

    ORDER BY OrderYear;

%%pyspark

# Read the sql view (created in the sql cell) into this pyspark cell

df2 = spark.sql("SELECT * FROM so2")

Translating between pySpark and SQL

Creating a pyspark query using pyspark.sql.functions is much more difficult and hard to read than sql,

but can be more flexible. Simple group and sum, min, max, avg, etc. can be simple, but the above query contains

a sum of a calculation, which I think is far more readable and obvious in SQL than in pyspark.

This simple "get where", SQL query...

    SELECT

        CustomerName,

        Email

    FROM salesorders

    WHERE Item = 'Road-250 Red, 52'

Could be written in pyspark as...

df = spark.read.table("salesorders")

customers = (

    df

    .filter(df['Item']=='Road-250 Red, 52')

    .select("CustomerName", "Email")

)

or if we don't want the df reference, we can import col from pyspark.sql.functions

from pyspark.sql.functions import col

customers = (

    spark.read.table("salesorders")

    .filter(col("Item") == "Road-250 Red, 52")

    .select("CustomerName", "Email")

)

Note: Use either Filter or Where, as they are the same function.

We can also use a pass-through sql query as mentioned earlier using spark.sql("query")

e.g. The following sql query demonstates were magic sql cells excel in readability.

%%sql

SELECT

    YEAR(OrderDate) AS OrderYear,

    SUM((UnitPrice * Quantity) + Tax) AS GrossRevenue

FROM salesorders

GROUP BY YEAR(OrderDate)

ORDER BY OrderYear;

can be written in pyspark as...

%%pyspark

from pyspark.sql.functions import year, sum, col

salesorders = spark.read.table("salesorders")

result = (

    salesorders

    .withColumn("OrderYear", year(col("OrderDate")))

    .groupBy("OrderYear")

    .agg(sum((col("UnitPrice") * col("Quantity")) + col("Tax")).alias("GrossRevenue"))

    .orderBy("OrderYear")

)

withColumn() adds a new named column and specifies how to calculate it.

groupBy().agg() is used when we need to apply multiple aggregate calculations within a group, which in this case is aliased.

pyspark.sql.functions.col() gets the value of the column.

https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.agg.html

https://sparkbyexamples.com/pyspark/pyspark-groupby-agg-aggregate-explained/