Module 24: Cheat Sheets

dbutils.fs.ls("/Volumes/main/dbx_course/raw")          # list files
dbutils.fs.cp(src, dst, recurse=True)                   # copy (also mv, rm, mkdirs, head, put)
dbutils.widgets.text("run_date", "2025-01-01")          # job parameter in a notebook
dbutils.widgets.get("run_date")
dbutils.secrets.get(scope="my_app", key="api_key")      # value is redacted in output
dbutils.notebook.run("./child", 600, {"p": "x"})        # run another notebook
dbutils.notebook.exit("done")                           # return a value
dbutils.jobs.taskValues.set(key="n", value=5)           # pass values between job tasks
dbutils.help()                                          # list everything

%pip install requests==2.32.3

dbutils.library.restartPython()    # restart Python so new packages load

from pyspark.sql import functions as F

df.select(
    F.col("order_id"),
    F.when(F.col("quantity") >= 5, "bulk").otherwise("single").alias("size"),
    F.date_trunc("month", "order_ts").alias("month"),
)

%sql
SELECT customer_id, order_id
FROM dbx_course.orders
QUALIFY row_number() OVER (PARTITION BY customer_id ORDER BY order_ts DESC) = 1

(df.filter("quantity > 0")
   .withColumn("revenue", F.col("quantity") * F.col("price"))
   .groupBy("customer_id").agg(F.sum("revenue").alias("total"))
   .orderBy(F.desc("total"))
   .limit(10)
   .show())

big.join(F.broadcast(small), "product_id", "left")         # force a broadcast
spark.table("orders").join(spark.table("customers").hint("broadcast"), "customer_id")

SELECT /*+ BROADCAST(c) */ * FROM orders o JOIN customers c USING (customer_id)

w = Window.partitionBy("customer_id").orderBy("order_ts")
w_run = w.rowsBetween(Window.unboundedPreceding, Window.currentRow)

df.select(
    "customer_id", "order_ts",
    F.row_number().over(w).alias("n"),
    F.lag("order_ts").over(w).alias("prev_ts"),
    F.sum("quantity").over(w_run).alias("running_qty"),
)

from delta.tables import DeltaTable

dt = DeltaTable.forName(spark, "dbx_course.customers")
dt.alias("t").merge(updates.alias("s"), "t.customer_id = s.customer_id") \
  .whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()

(spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .option("cloudFiles.schemaLocation", "/Volumes/main/dbx_course/raw/_schema/orders")
      .load("/Volumes/main/dbx_course/raw/landing/orders"))

(df.withWatermark("event_ts", "10 minutes")
   .groupBy(F.window("event_ts", "5 minutes"), "event_type").count()
   .writeStream
   .outputMode("append")
   .option("checkpointLocation", "/Volumes/main/dbx_course/raw/_cp/by_type")
   .trigger(availableNow=True)
   .toTable("events_by_type"))

from pyspark import pipelines as dp
from pyspark.sql import functions as F

@dp.table(comment="Clean orders")
@dp.expect_or_drop("positive_qty", "quantity > 0")
def orders_clean():
    return spark.readStream.table("orders_bronze").withColumn("status", F.lower("status"))

CREATE OR REFRESH STREAMING TABLE orders_bronze
AS SELECT * FROM STREAM read_files('/Volumes/main/dbx_course/raw/landing', format => 'json');

databricks auth login --host https://<workspace-url>     # sign in (OAuth)
databricks current-user me                               # who am I?
databricks workspace list /                              # list workspace paths
databricks workspace export /Users/me/nb --format SOURCE
databricks fs ls dbfs:/Volumes/<catalog>/<schema>/<volume>
databricks jobs list
databricks jobs run-now <job-id>
databricks catalogs list
databricks bundle init | validate | deploy -t dev | run <job> -t dev | destroy -t dev
databricks secrets create-scope my_app
databricks --help