Examples gallery

DuckDB runtimeDownload PDF (2.8 MB)

The examples use fictional data for a retail company, Acme Retail. Connection names (sales_s3, orders_pg, lakehouse_main, ...) are placeholders for your own.

Example 1: Partitioned daily export#

Aggregate all order files and write one Parquet folder per region, ready for downstream readers.

pythonfrom GZ.DuckDBUtils import duckSession

session = duckSession()

session.readS3File(
    connName="sales_s3",
    s3FullUrl="s3://acme-retail-data/raw/orders/*.parquet",
    targetDuckdbTable="orders",
)

# The S3 call above configured access, so plain SQL can COPY to S3 too.
session.query("""
    COPY (
        SELECT region,
               CAST(ordered_at AS DATE) AS order_date,
               count(*)                 AS orders,
               round(sum(amount), 2)    AS revenue
        FROM orders
        WHERE status <> 'returned'
        GROUP BY ALL
    )
    TO 's3://acme-retail-data/curated/revenue_by_region'
    (FORMAT PARQUET, PARTITION_BY (region), OVERWRITE_OR_IGNORE)
""")

check = session.query(
    "SELECT count(*) AS files FROM glob("
    "'s3://acme-retail-data/curated/revenue_by_region/*/*.parquet')",
    returnFormat="pandas",
)
print(check)

Example 2: Rolling averages with window functions#

DuckDB's analytic SQL makes time-series features a one-liner. This computes a 7-day moving average of revenue and ranks days within each region.

pythonfrom GZ.DuckDBUtils import duckSession

session = duckSession()
session.readS3File(
    "sales_s3", s3FullUrl="s3://acme-retail-data/raw/orders/*.parquet",
    targetDuckdbTable="orders",
)

outputDF = session.query("""
    WITH daily AS (
        SELECT region,
               CAST(ordered_at AS DATE) AS day,
               sum(amount)              AS revenue
        FROM orders
        GROUP BY region, day
    )
    SELECT region, day, round(revenue, 2) AS revenue,
           round(avg(revenue) OVER (
               PARTITION BY region ORDER BY day
               ROWS BETWEEN 6 PRECEDING AND CURRENT ROW), 2) AS revenue_7d_avg,
           rank() OVER (PARTITION BY region ORDER BY revenue DESC) AS best_day_rank
    FROM daily
    ORDER BY region, day
""")

Example 3: Flatten JSON events#

Semi-structured JSON is a first-class citizen. This reads newline-delimited event files and extracts nested fields without leaving SQL.

pythonfrom GZ.DuckDBUtils import duckSession

session = duckSession()
session.readS3File(
    connName="sales_s3",
    bucketName="acme-retail-data",
    filePath="raw/events/events.json",
    fileType="json",
    targetDuckdbTable="events",
)

outputDF = session.query("""
    SELECT user_id,
           event,
           props.page                AS page,
           CAST(props.duration_ms AS INTEGER) AS duration_ms,
           CAST(ts AS TIMESTAMP)     AS ts
    FROM events
    WHERE event IN ('page_view', 'checkout')
    ORDER BY ts
""")

Example 4: Churn scoring with scikit-learn#

Train a model on features computed in SQL, score every customer and write the scores back. Use the DuckDB scikit-learn (or PyTorch or TensorFlow) variant, or any runtime that includes scikit-learn.

pythonfrom sklearn.linear_model import LogisticRegression

from GZ.DuckDBUtils import duckSession

session = duckSession()
session.readS3File(
    "sales_s3", s3FullUrl="s3://acme-retail-data/raw/orders/*.parquet",
    targetDuckdbTable="orders",
)

features = session.query("""
    SELECT customer_id,
           count(*)                                   AS n_orders,
           avg(amount)                                AS avg_amount,
           max(amount)                                AS max_amount,
           max(CASE WHEN status = 'returned' THEN 1 ELSE 0 END) AS had_return
    FROM orders
    GROUP BY customer_id
""", returnFormat="pandas")

x = features[["n_orders", "avg_amount", "max_amount"]]
y = features["had_return"]
model = LogisticRegression(max_iter=500).fit(x, y)

features["return_risk"] = model.predict_proba(x)[:, 1].round(4)
session.register("scores", features[["customer_id", "return_risk"]])

session.writeS3File(
    connName="sales_s3",
    s3FullUrl="s3://acme-retail-data/curated/return_risk.parquet",
    sourceTable="scores",
)
outputDF = session.query("SELECT * FROM scores ORDER BY return_risk DESC LIMIT 10")

Example 5: Deduplicate and validate before loading#

A light quality gate that runs entirely in DuckDB: reject malformed rows, keep the latest record per order and report what was dropped.

pythonfrom GZ.DuckDBUtils import duckSession

session = duckSession()
session.readS3File(
    "sales_s3", s3FullUrl="s3://acme-retail-data/raw/orders/*.parquet",
    targetDuckdbTable="raw_orders",
)

session.query("""
    CREATE TABLE clean_orders AS
    SELECT * EXCLUDE (rn)
    FROM (
        SELECT *, row_number() OVER (
                   PARTITION BY order_id ORDER BY ordered_at DESC) AS rn
        FROM raw_orders
        WHERE amount IS NOT NULL AND amount >= 0 AND region IS NOT NULL
    )
    WHERE rn = 1
""")

stats = session.query("""
    SELECT (SELECT count(*) FROM raw_orders)   AS raw_rows,
           (SELECT count(*) FROM clean_orders) AS clean_rows
""", returnFormat="pandas")
print(stats.to_dict("records")[0])

session.writeS3File(
    "sales_s3", s3FullUrl="s3://acme-retail-data/curated/clean_orders.parquet",
    sourceTable="clean_orders",
)

Example 6: Customer 360 across lakehouse, PostgreSQL and S3#

One query joins three systems. The lakehouse read is governed for the calling user; the PostgreSQL and S3 reads stream into DuckDB.

pythonfrom GZ.DuckDBUtils import duckSession

session = duckSession(siteName="acme")

orders = session.readGzLakehouse(
    connName="lakehouse_main",
    query="""
        SELECT customer_id, count(*) AS orders, sum(amount) AS revenue
        FROM retail.orders GROUP BY customer_id
    """,
)
profiles = session.executeJdbc(
    connName="orders_pg",
    sqlQuery="SELECT customer_id, segment, signup_date FROM public.customers",
)
tiers = session.readS3File(
    connName="sales_s3",
    bucketName="acme-retail-data",
    filePath="reference/loyalty_tiers.csv",
    fileType="csv",
)

session.register("orders", orders)
session.register("profiles", profiles)
session.register("tiers", tiers)

outputDF = session.query("""
    SELECT p.customer_id, p.segment, t.tier,
           o.orders, round(o.revenue, 2) AS revenue
    FROM profiles p
    LEFT JOIN orders o USING (customer_id)
    LEFT JOIN tiers t ON o.revenue BETWEEN t.min_revenue AND t.max_revenue
    ORDER BY revenue DESC NULLS LAST
""")

Example 7: Incremental load from Snowflake#

Pull only the rows newer than the last load, append them to a Parquet history file set and report the row count.

pythonfrom GZ.DuckDBUtils import duckSession

session = duckSession()

session.readS3File(
    connName="sales_s3",
    s3FullUrl="s3://acme-retail-data/history/orders/*.parquet",
    targetDuckdbTable="history",
)
last = session.query(
    "SELECT coalesce(max(order_ts), TIMESTAMP '1970-01-01') AS w FROM history",
    returnFormat="pandas",
)["w"].iloc[0]

rows = session.readSnowflake(
    conn_name="wh_snowflake",
    sql_query=f"""
        SELECT order_id, customer_id, amount, order_ts
        FROM ANALYTICS.PUBLIC.ORDERS
        WHERE order_ts > '{last:%Y-%m-%d %H:%M:%S}'
    """,
    target_duckdb_table="orders_new",
    returnFormat="row_count",
)
print(f"loaded {rows} new orders")

if rows:
    session.writeS3File(
        connName="sales_s3",
        s3FullUrl=(
            "s3://acme-retail-data/history/orders/"
            f"batch_{last:%Y%m%d%H%M%S}.parquet"
        ),
        sourceTable="orders_new",
    )

Example 8: Classify support tickets with an LLM#

Read tickets from PostgreSQL, add a sentiment label from an LLM connection and save the result. continueOnError=True keeps one bad row from failing the run.

pythonfrom GZ.DuckDBUtils import duckSession

session = duckSession()

tickets = session.executeJdbc(
    connName="orders_pg",
    sqlQuery="SELECT ticket_id, body FROM support.tickets WHERE status = 'open'",
    returnFormat="pandas",
)

labelled = session.complete(
    connName="llm_openai",
    df=tickets,
    promptTemplate=(
        "Classify the sentiment of this support ticket as positive, "
        "neutral or negative. Reply with one word.\n\n{body}"
    ),
    outputColumn="sentiment",
    continueOnError=True,
    timeout=60,
)
session.register("labelled", labelled)

session.writeS3File(
    connName="sales_s3",
    s3FullUrl="s3://acme-retail-data/curated/ticket_sentiment.parquet",
    sourceQuery="""
        SELECT ticket_id, lower(trim(sentiment)) AS sentiment,
               sentiment_status, sentiment_total_tokens
        FROM labelled
    """,
)

Example 9: Ingest a REST API#

Load paginated JSON from an API connection, keep the fields you need and write them as CSV.

pythonfrom GZ.DuckDBUtils import duckSession

session = duckSession()

session.executeREST(
    connName="support_api",
    endpoint="/v2/tickets",
    params={"status": "open", "per_page": 100},
    resultPath="data.items",
    maxPages=20,
    maxRecords=2000,
    targetDuckdbTable="api_tickets",
)

session.writeS3File(
    connName="sales_s3",
    s3FullUrl="s3://acme-retail-data/raw/tickets/open.csv",
    fileType="csv",
    sourceQuery="SELECT id, subject, created_at FROM api_tickets",
    writeOptions={"header": True},
)

Example 10: Geospatial distance query#

The spatial extension is installed. This finds the stores within 50 km of a point.

pythonfrom GZ.DuckDBUtils import duckSession

session = duckSession()
session.query("LOAD spatial")

session.query("""
    CREATE TABLE stores AS
    SELECT * FROM (VALUES
        ('Berlin',  52.5200, 13.4050),
        ('Potsdam', 52.3906, 13.0645),
        ('Hamburg', 53.5511,  9.9937)
    ) AS t(store, lat, lon)
""")

outputDF = session.query("""
    SELECT store,
           round(ST_Distance_Sphere(
               ST_Point(lon, lat), ST_Point(13.4050, 52.5200)) / 1000, 1)
               AS km_from_berlin
    FROM stores
    WHERE ST_Distance_Sphere(
            ST_Point(lon, lat), ST_Point(13.4050, 52.5200)) <= 50000
    ORDER BY km_from_berlin
""")

Example 11: Extract policy fields from PDFs with an Agent#

Parse a folder of PDFs and ask an approved Agent to fill your JSON schema for each one. The Agent's response schema, configured in Agent Builder, decides the output shape.

pythonfrom GZ.DuckDBUtils import duckSession

session = duckSession()
try:
    result = session.invokeAgentPDFCorpus(
        connName="agents",
        agentId="00000000-0000-4000-8000-000000000001",
        message="Extract the policy number, holder name and effective date.",
        pdfSourceConnName="sales_s3",
        pdfSourcePrefix="incoming/policies/",
        pdfSinkConnName="sales_s3",
        pdfSinkPrefix="agent-results/policies",
        summaryWorkers=8,
    )
    print(result["status"], result["outputMode"])
    print(result["resultsUri"])
    print(result["manifestUri"])
finally:
    session.close()

Example 12: A two-step pipeline with executeJob#

Put the heavy preparation in a sub-job and let the parent job orchestrate and notify.

pythonfrom GZ.JobUtils import executeJob
from GZ.SendEmail import sendEmail

results = executeJob(["load_orders", "build_summary"])

sendEmail(
    subject="Retail pipeline finished",
    body=f"Completed sub-jobs: {', '.join(results)}",
    to_email="data-team@example.com",
    from_email="noreply@example.com",
)