Examples gallery

Spark runtimeDownload PDF (3.0 MB)

All examples use fictional names and data. Replace connection names (acme-*) with your own. Each example is a complete job script.

Daily revenue rollup in the lakehouse#

Reads yesterday's orders from a lakehouse table, aggregates them by region and appends the result to a summary table. The aggregate is also shown as the job preview.

python
from GZ.SparkUtils import sparkSession
from pyspark.sql import functions as F

spark = sparkSession("DailyRevenueRollup", siteName="acme")

orders = spark.executeQuery(
    connName="acme-lakehouse",
    sqlQuery="""
        SELECT region, product, quantity, unit_price
        FROM sales.orders
        WHERE order_date = date_sub(current_date(), 1)
    """,
)

daily = (
    orders.withColumn("revenue", F.col("quantity") * F.col("unit_price"))
    .groupBy("region")
    .agg(
        F.sum("revenue").alias("revenue"),
        F.countDistinct("product").alias("products"),
    )
    .withColumn("report_date", F.date_sub(F.current_date(), 1))
)

spark.writeGzLakehouse(
    df=daily,
    connName="acme-lakehouse",
    tableName="revenue_daily",
    database="sales",
    mode="append",
)

outputDF = daily.orderBy(F.desc("revenue"))

Incremental load from PostgreSQL to Parquet#

Pulls rows changed in the last day, stamps them with the run's name and start time through placeholders, and appends them to Parquet in S3.

python
from GZ.SparkUtils import sparkSession
from pyspark.sql import functions as F

spark = sparkSession("IncrementalCustomers")

run_job = {JOBNM}.strip("'")
run_started = {STRTTM}.strip("'")

changes = spark.readJDBC(
    connName="acme-postgres",
    query="SELECT customer_id, email, country, updated_at "
    "FROM crm.customers WHERE updated_at >= now() - interval '1 day'",
)

landed = (
    changes.withColumn("loaded_by", F.lit(run_job))
    .withColumn("loaded_at", F.lit(run_started))
)
spark.writeS3(
    df=landed,
    connName="acme-s3",
    filePath="landing/crm/customers",
    format="parquet",
    mode="append",
)

print("Loaded", landed.count(), "changed customers")
outputDF = landed.groupBy("country").count()

The placeholders are substituted as quoted string literals, so .strip("'") removes the quotes that are meant for SQL.

Deduplicate and enrich with window functions#

Keeps the latest record per customer and joins a country reference table read from CSV.

python
from GZ.SparkUtils import sparkSession
from pyspark.sql import Window
from pyspark.sql import functions as F

spark = sparkSession("DedupeCustomers")

customers = spark.readS3(
    connName="acme-s3",
    fileName="landing/crm/customers_raw/",
    format="csv",
    options={"header": "true", "inferSchema": "true"},
)
countries = spark.readS3(
    connName="acme-s3",
    fileName="reference/countries.csv",
    format="csv",
    options={"header": "true"},
)

latest = Window.partitionBy("customer_id").orderBy(F.desc("updated_at"))
deduped = (
    customers.withColumn("rank", F.row_number().over(latest))
    .where(F.col("rank") == 1)
    .drop("rank")
    .join(F.broadcast(countries), on="country_code", how="left")
)

spark.writeS3(
    df=deduped,
    connName="acme-s3",
    filePath="curated/customers",
    format="parquet",
    mode="overwrite",
)
outputDF = deduped

Stream Kafka events into an Iceberg table#

Consumes a topic and appends micro-batches to a lakehouse table. With the availableNow trigger the job processes everything that has arrived and then finishes, which suits scheduled runs.

python
from GZ.SparkUtils import sparkSession
from pyspark.sql import functions as F

spark = sparkSession("OrderEventsStream", siteName="acme")

raw = spark.readKafka(
    connName="acme-kafka",
    topic="orders.created",
    startingOffsets="earliest",
)

events = raw.select(
    F.col("key").cast("string").alias("order_key"),
    F.get_json_object(F.col("value").cast("string"), "$.region").alias("region"),
    F.get_json_object(F.col("value").cast("string"), "$.total")
    .cast("double")
    .alias("total"),
    F.col("timestamp").alias("event_time"),
)

query = spark.writeGzLakehouse(
    df=events,
    connName="acme-lakehouse",
    tableName="order_events",
    database="sales",
    streaming=True,
    checkpointLocation="s3://acme-data/checkpoints/order_events",
    outputMode="append",
    trigger={"availableNow": True},
)
query.awaitTermination()

Create the table first

The target sales.order_events table must already exist with the partition specification you want, because the streaming writer does not apply partitionBy.

Classify support tickets with an LLM#

Calls an LLM connection once per row, keeps going when individual rows fail, and saves successes and failures separately.

python
from GZ.SparkUtils import sparkSession
from pyspark.sql import functions as F

spark = sparkSession("TicketTriage")

tickets = spark.executeQuery(
    connName="acme-lakehouse",
    sqlQuery="SELECT ticket_id, subject, body FROM support.tickets LIMIT 500",
    siteName="acme",
)

scored = spark.complete(
    connName="acme-openai",
    df=tickets,
    promptTemplate=(
        "Classify this support ticket as billing, technical or other. "
        "Reply with one word.\nSubject: {{subject}}\nBody: {{body}}"
    ),
    systemPrompt="You are a precise ticket router.",
    outputColumn="category",
    maxTokens=5,
    continueOnError=True,
)

good = scored.where(F.col("category_status") == "success")
bad = scored.where(F.col("category_status") == "error")

spark.writeS3(good, "acme-s3", "triage/ok", "parquet", "overwrite")
spark.writeS3(bad, "acme-s3", "triage/errors", "parquet", "overwrite")

outputDF = good.groupBy(F.lower(F.trim("category")).alias("category")).count()

Train and apply a scikit-learn model#

Trains on a sample collected to the driver, then scores the full DataFrame in parallel with a pandas UDF. Run it on the scikit-learn (or another ML) variant, or on the base runtime, since scikit-learn is part of the base runtime.

python
import pandas as pd
from GZ.SparkUtils import sparkSession
from pyspark.sql import functions as F
from pyspark.sql.functions import pandas_udf
from sklearn.ensemble import GradientBoostingClassifier

spark = sparkSession("ChurnModel")

features = ["tenure_months", "monthly_spend", "support_tickets"]
customers = spark.executeQuery(
    connName="acme-lakehouse",
    sqlQuery="SELECT customer_id, tenure_months, monthly_spend, "
    "support_tickets, churned FROM crm.customer_features",
    siteName="acme",
)

sample = customers.sample(fraction=0.2, seed=7).limit(50_000).toPandas()
model = GradientBoostingClassifier(random_state=7)
model.fit(sample[features], sample["churned"])


@pandas_udf("double")
def churn_probability(*cols: pd.Series) -> pd.Series:
    frame = pd.concat(cols, axis=1)
    frame.columns = features
    return pd.Series(model.predict_proba(frame)[:, 1])


scored = customers.withColumn(
    "churn_probability", churn_probability(*[F.col(c) for c in features])
)
outputDF = scored.select("customer_id", "churn_probability").orderBy(
    F.desc("churn_probability")
)

Load a warehouse query into the lakehouse in bulk#

Verifies the lakehouse route, lets Snowflake unload the rows directly into the table, then counts them.

python
from GZ.SparkUtils import sparkSession

spark = sparkSession("OrdersBulkLoad", siteName="acme")

check = spark.verifyLakehouseConnection("acme-lakehouse", siteName="acme")
print("Routing to warehouse", check["warehouse"], "at site", check["site_used"])

summary = spark.bulkLoadGzLakehouseFromSnowflake(
    connName="acme-snowflake",
    query="SELECT * FROM analytics.orders WHERE order_date >= '2026-01-01'",
    lakehouseConn="acme-lakehouse",
    tableName="orders_history",
    database="sales",
    mode="overwrite",
    warehouseSize="X-Large",
    siteName="acme",
)
print("Bulk load summary:", summary)

outputDF = spark.executeQuery(
    connName="acme-lakehouse",
    sqlQuery="SELECT COUNT(*) AS loaded_rows FROM sales.orders_history",
)

Extract structured data from a folder of PDFs with an Agent#

Runs an approved Groundzero Agent over every PDF in a prefix and reads the results back as a DataFrame.

python
from GZ.SparkUtils import sparkSession

spark = sparkSession("InvoiceExtraction")

result = spark.invokeAgentPDFCorpus(
    connName="acme-agents",
    agentId="123e4567-e89b-42d3-a456-426614174000",
    message="Extract the invoice number, vendor and total into the schema.",
    pdfSourceConnName="acme-s3",
    pdfSourcePrefix="incoming/invoices/",
    pdfSinkConnName="acme-s3",
    pdfSinkPrefix="agent-pdf-results/invoices",
    summaryWorkers=8,
)
print(result["status"], result["outputMode"])
print("Processed", result["processedDocumentCount"], "documents")

outputDF = spark.readS3(
    connName="acme-s3",
    fileName="agent-pdf-results/invoices/",
    format="json",
)

Quality gate with notification#

Runs checks on a table, writes the HTML report, and mails the team when a column is out of range.

python
from GZ.SendEmail import sendEmail
from GZ.SparkUtils import sparkSession
from qualipilot import CheckConfig, QualipilotConfig
from qualipilot.models.config import ColumnRange

spark = sparkSession("OrdersQualityGate")

orders = spark.executeQuery(
    connName="acme-lakehouse",
    sqlQuery="SELECT order_id, customer_email, amount FROM sales.orders",
    siteName="acme",
)
config = QualipilotConfig(
    checks=CheckConfig(
        required_columns=["order_id", "customer_email", "amount"],
        column_ranges={"amount": ColumnRange(min=0, max=100_000)},
    )
)
report = spark.runDataQualityChecks(
    df=orders, connectionName="acme-openai", config=config
)
spark.saveDataQualityReport(report, "reports/orders_quality.html", "html")

negative = orders.where("amount < 0").count()
if negative:
    sendEmail(
        subject="Orders quality gate failed",
        body=f"{negative} orders have a negative amount.",
        to_email="data-team@example.com",
        from_email="alerts@example.com",
    )
outputDF = orders.where("amount < 0")

Orchestrate sub-jobs#

A parent job runs two sub-jobs in order and then reports their outputs.

python
from GZ.JobUtils import executeJob
from GZ.SparkUtils import sparkSession

spark = sparkSession("NightlyPipeline")

results = executeJob(["load_orders", "build_marts"])
for name, output in results.items():
    print("Finished", name, "with output", output)

outputDF = spark.createDataFrame(
    [(name,) for name in results], ["completed_job"]
)