2 Databricks Demo 2 – GitHub File Ingestion & Delta Table Creation

Recap of Previous Session

  • Defined Databricks as a “Data Intelligence Platform” and walked through high-level architecture (workspaces, clusters/compute, Unity Catalog, volumes, tables).
  • Completed a first hands-on use case:
    • Uploaded a local CSV into a volume (my_volume/input/…).
    • Created a table on top of it and ran a few exploratory queries.
  • Emphasised that we are still familiarising ourselves with the platform UI & concepts—core Spark/PySpark theory will follow in later lectures.

New Use Case – Analyse a File Stored on the Web (GitHub)

  • Goal: ingest a public CSV directly from GitHub, land it in a volume, add a calculated column, and persist the result as a Delta table for SQL analysis.
  • Major steps:
    1. Copy file from GitHub URL → Unity Catalog volume (my_volume/input).
    2. Read the file into a Spark DataFrame with correct schema.
    3. Derive a new timestamp column (last_updated).
    4. Persist to a managed Delta table (dev.demo.git_sales).
    5. Validate with SQL and inspect storage-level artefacts.

Environment Preparation

  • Switched to catalog dev, schema demo (created yesterday).
  • Confirmed existing objects:
    • 2 tables (sales, sales_report_7370) + one volume (my_volume).
  • Created a new notebook named “Second Demo Use Case”; default language chosen as Python (can still embed SQL/R/Scala cells via ‘magic commands’).
  • Spun up a compute cluster before running cells.
  • From the Catalog UI → Details → observed physical storage path:
    • Container: unity-catalog
    • Storage account: kubixdev
    • Volume path: …/volumes/my_volume with sub-folder input/.
  • Browsed the same path in Azure Storage Explorer to see that the volume already contains two CSV files.

dbutils.fs – File-System Utilities

  • dbutils.help() → shows helper libraries (fs, jobs, libraries, secrets, …).
  • dbutils.fs.help("cp") → reveals the cp method signature (from, to) for cross-filesystem copy.
  • Commands used:
  source_path  = "https://raw.githubusercontent.com/<repo>/datasets/dataflow_transformations/select/sales_orders.csv"
  target_path  = "/Volumes/dev/demo/my_volume/input/git_sales_orders.csv"

  dbutils.fs.cp(source_path, target_path)
  • Verified before & after state with:
  display(dbutils.fs.ls('/Volumes/dev/demo/my_volume/input'))

Reading CSV into a DataFrame

  • Core API: spark.read.format('csv').
  • Important options:
    • header='true' → treat first row as column names.
    • inferSchema='true' → sample data and auto-detect column types.
  • Example (split across lines for readability):
  sales_df = (
      spark.read
           .format('csv')
           .option('header', 'true')
           .option('inferSchema', 'true')
           .load('/Volumes/dev/demo/my_volume/input/git_sales_orders.csv')
  )

Inspecting a DataFrame

  • display(sales_df) – rich grid in notebook.
  • sales_df.show() – console-style output.
  • sales_df.head(5) – returns first 5 rows (Python list of Row objects).
  • Schema introspection:
    • sales_df.printSchema() – tree view (root -> column:type (nullable)).
    • sales_df.schema or sales_df.dtypes – programmatic metadata.

DataFrame Immutability & Re-assignment

  • Spark DataFrames are immutable; transformations return new DF instances.
  • Pattern used: either assign to a new variable or re-use the same name (conceptually drops & re-creates the variable).
  df2 = df1.withColumn(...)
  # or
  df1 = df1.withColumn(...)

Deriving a New Column (withColumn)

  • Need: add last_updated that captures current timestamp at load time.
  • Imported helper functions once per notebook:
  from pyspark.sql.functions import *   # current_timestamp, col, etc.
  • Transformation:
  sales_df = sales_df.withColumn('last_updated', current_timestamp())
  • Example of a calculated numeric column (price × quantity):
  sales_df = sales_df.withColumn(
      'sales_amount',
      col('price') * col('quantity_sold')
  )

Writing to a Delta Table

  • Unity Catalog uses a three-level namespace: catalog.schema.table\text{catalog.schema.table} (e.g.
    dev.demo.git_sales).
  • API used:
  sales_df.write
          .mode('overwrite')   # optional; default for new tables
          .format('delta')     # explicit; implicit if omitted
          .saveAsTable('dev.demo.git_sales')
  • Confirmed creation:
    • Catalog explorer refresh shows new table.
    • Physical artefacts in storage:
    • GUID folder under …/Tables/ containing Parquet data files and _delta_log/.
  • Validation queries:
  %sql
  SELECT * FROM dev.demo.git_sales;

  DESCRIBE TABLE EXTENDED dev.demo.git_sales; -- shows format = delta, location, stats

Running SQL Two Ways

  1. Magic Command: prepend cell with %sql (or %%sql in classic notebooks).
  2. Programmatic: spark.sql("<SQL>") inside Python; returns a DataFrame.
   agg_df = spark.sql(
       """
       SELECT customer_name, SUM(price * quantity_sold) AS revenue
       FROM dev.demo.git_sales
       GROUP BY customer_name
       """
   )
   display(agg_df)

Dynamic SQL with Python f-Strings

  • Advantage of spark.sql() = SQL text is just a Python string → can be parameterised.
  table_name = 'dev.demo.git_sales'
  query = f"SELECT COUNT(*) FROM {table_name}"
  spark.sql(query).show()
  • Multi-line strings use triple quotes """ ... """ and can embed variables via {}.

Dropping & Re-running End-to-End

  • Added a safeguard cell before table creation:
  %sql
  DROP TABLE IF EXISTS dev.demo.git_sales;
  • Notebook toolbar → Run All executes every cell sequentially (copy → transform → write).

Delta Table Concepts Highlighted

  • Delta format = Parquet files + transaction log directory _delta_log.
  • Guarantees ACID, schema enforcement, versioning—topic for future deep dive.
  • DESCRIBE TABLE EXTENDED revealed key metadata: type = MANAGED, provider = delta, physical location, stats.

DataFrame vs. Table – Mental Model

  • DataFrame: runtime, in-memory (or cached) variable used exclusively for computation; disappears when notebook/cluster terminates.
  • Delta Table: persistent storage object (data lives in Azure Data Lake Gen2 under the GUID folder); survives notebook deletion.

Notebook Lifecycle & Best-Practice Points

  • Import statements (from pyspark.sql.functions import *) are notebook-scoped; execute once per new notebook.
  • Keep code cells concise; break long method chains with back-slash continuation for readability.
  • Use Markdown cells (%md) for documentation; execution-neutral.
  • Execution order matters—ensure dependencies appear above consumer cells.

Quick Command Reference

  • File copy/list: dbutils.fs.cp, dbutils.fs.ls.
  • Read CSV: spark.read.format('csv').option(...).load(path).
  • Inspect DF: display(df), df.show(), df.printSchema().
  • Add column: df.withColumn(newName, expr).
  • Write Delta: df.write.mode('overwrite').saveAsTable(fullName).
  • SQL cell: %sql <query>.
  • Programmatic SQL: spark.sql(<string>).
  • Drop table: DROP TABLE IF EXISTS catalog.schema.table.

Connections & Broader Context

  • Mirrors Azure Data Factory’s Derived Column transformation, but here implemented via PySpark and DataFrames.
  • Demonstrates Databricks’ lakehouse paradigm: ingest raw file → transform in code → persist as Delta for BI/SQL.
  • Reinforces the importance of Unity Catalog for governance (three-level namespacing) and Delta for reliability.
  • Ethical/practical implication: using public GitHub data highlights need for data provenance and license compliance when ingesting external datasets.

What’s Next

  • Deep dive into Spark architecture (driver, executors, partitions).
  • Formal introduction to PySpark API (DataFrame vs. Dataset, transformations vs. actions).
  • Delta Lake features: time-travel, schema evolution, MERGE (upserts).