databricks_dlt_medallion_bronze_silver_with_expectations.py

1import dlt
2from pyspark.sql.functions import *
3
4# Define the bronze table to ingest raw data
5@dlt.table(
6  comment="The raw terminology data, ingested from samples."
7)
8def wikipedia_raw():
9  return (
10    spark.readStream.format("cloudFiles")
11      .option("cloudFiles.format", "json")
12      .load("/databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed-json/2015_03_04_20_000.json")
13  )
14
15# Define the silver table to clean and refine the data
16@dlt.table(
17  comment="Cleaning the raw data by filtering nulls and selecting specific columns."
18)
19@dlt.expect_or_drop("valid_current_page_title", "curr_title IS NOT NULL")
20def wikipedia_prepared():
21  return (
22    dlt.read("wikipedia_raw")
23      .select("curr_title", "curr_id", "prev_title", "prev_id", "type", "n")
24  )