Introduction

This cookbook shows two production‑ready patterns to land live, structured web data from Nimble Way into your analytics stack and organize it with a Medallion (Bronze/Silver/Gold) layout:

Snowflake: ingest with Snowpipe Streaming (via Kafka) into a Bronze table, then parse to Silver/Gold.
Databricks: ingest with Auto Loader and Delta Live Tables (DLT) to build Bronze/Silver/Gold pipelines.

Nimble delivers clean, real‑time web data and streams directly into destinations like Snowflake, Databricks, S3, and GCS, with governance and compliance by design.

Architecture at a glance

Flow	Ingest	Bronze	Silver	Gold
Snowflake	Kafka → Snowpipe Streaming	VARIANT (raw events)	Parsed, typed view/table	Business‑ready marts
Databricks	S3/GCS → Auto Loader (DLT)	Raw JSON Delta table	Normalized Delta table	Aggregations/feature tables

References: Nimble delivers governed, analysis‑ready data with native destinations (Snowflake, Databricks) and quality/observability.

Prerequisites

Nimble account and API key; configured Online Pipeline or API call returning structured JSON.
Kafka cluster (for Snowpipe Streaming path) and access to deploy a Kafka Connect sink.
Snowflake account with role to create database/schema/tables and a service user for ingestion.
Databricks workspace with DLT entitlement; an S3 (or GCS) bucket for Bronze landing and checkpoints.
AWS IAM role for Databricks to read/write the bucket (IAM snippets below).
Compliance: Nimble is built for CCPA/GDPR/SOC2 and ethical collection.

Sample event schemas (JSON)

Below are compact, production‑friendly schemas you can use across both targets.

Retail pricing event (JSON):

{
 "_meta": {"source":"amazon","country":"US","collected_at":"2025-01-01T12:34:56Z"},
 "product": {"id":"B0ABC12345","title":"Widget Pro 64GB","brand":"Acme"},
 "offer": {"price":199.99,"currency":"USD","in_stock":true,"seller":"Acme Store","delivery":"2-day"},
 "serp": {"keyword":"widget pro","rank":3},
 "page": {"url":"(product URL)"}
}

SERP observation (JSON):

{
 "_meta": {"engine":"google","country":"US","collected_at":"2025-01-01T12:35:12Z"},
 "query":"best running shoes women",
 "position":1,
 "type":"shopping",
 "domain":"example-retailer.com",
 "price":89.99,
 "currency":"USD"
}

Part A — Snowflake with Snowpipe Streaming (Kafka)

This pattern uses Kafka → Snowpipe Streaming to land Nimble events into a Bronze VARIANT table, then projects typed Silver/Gold.

1) Snowflake DDL (Bronze/Silver/Gold)

Run as a Snowflake role with CREATE privileges.

-- Bronze (raw events)
CREATE DATABASE IF NOT EXISTS NIMBLE_DEMO;
CREATE SCHEMA IF NOT EXISTS NIMBLE_DEMO.RETAIL;
CREATE OR REPLACE TABLE NIMBLE_DEMO.RETAIL.PRICING_BRONZE (
 EVENT VARIANT,
 SRC_TS TIMESTAMP_LTZ DEFAULT CURRENT_TIMESTAMP()
);

-- Silver (typed projection as a view)
CREATE OR REPLACE VIEW NIMBLE_DEMO.RETAIL.PRICING_SILVER AS
SELECT
 EVENT:_meta:source::STRING AS source,
 EVENT:_meta:country::STRING AS country,
 TO_TIMESTAMP_NTZ(EVENT:_meta:collected_at::STRING) AS collected_at,
 EVENT:product:id::STRING AS product_id,
 EVENT:product:title::STRING AS title,
 EVENT:offer:price::FLOAT AS price,
 EVENT:offer:currency::STRING AS currency,
 EVENT:offer:in_stock::BOOLEAN AS in_stock,
 EVENT:offer:seller::STRING AS seller,
 EVENT:serp:rank::NUMBER AS serp_rank,
 EVENT:page:url::STRING AS url
FROM NIMBLE_DEMO.RETAIL.PRICING_BRONZE;

-- Gold (latest price per product/source)
CREATE OR REPLACE VIEW NIMBLE_DEMO.RETAIL.PRICING_GOLD AS
SELECT
 product_id,
 source,
 ANY_VALUE(title) AS title,
 LAST_VALUE(price) IGNORE NULLS WITHIN GROUP (ORDER BY collected_at) AS current_price,
 ANY_VALUE(currency) AS currency,
 MAX(collected_at) AS last_seen
FROM NIMBLE_DEMO.RETAIL.PRICING_SILVER
GROUP BY product_id, source;

2) Kafka Connect sink (Snowpipe Streaming) config

Deploy a Kafka Connect sink with the Snowflake connector. Example connector spec (JSON):

{
 "name": "sf-nimble-pricing-snowpipe-streaming",
 "config": {
 "connector.class": "com.snowflake.kafka.connector. SnowflakeSinkConnector",
 "tasks.max": "4",
 "topics": "nimble.pricing",

 "snowflake.url.name": "<ACCOUNT_IDENTIFIER>.snowflakecomputing.com",
 "snowflake.user.name": "KAFKA_CONNECTOR",
 "snowflake.private.key": "<BASE64_PK>",
 "snowflake.private.key.passphrase": "<PASSPHRASE>",
 "snowflake.database.name": "NIMBLE_DEMO",
 "snowflake.schema.name": "RETAIL",
 "snowflake.role.name": "NIMBLE_INGEST_ROLE",

 "snowflake.ingestion.method": "SNOWPIPE_STREAMING",

 "buffer.flush.time": "10",
 "buffer.count.records": "5000",
 "behavior.on.null.values": "IGNORE",
 "key.converter": "org.apache.kafka.connect.storage. StringConverter",
 "value.converter": "org.apache.kafka.connect.json. JsonConverter",
 "value.converter.schemas.enable": "false"
 }
}

Connector will write the topic payload into PRICING_BRONZE.EVENT as VARIANT rows.

3) Python producer: stream Nimble events to Kafka

Example producer that calls Nimble and writes to Kafka. Replace placeholders as noted.

import os, json, time, requests
from kafka import KafkaProducer

NIMBLE_API_URL = os.environ.get("NIMBLE_API_URL")

# e.g., https://api.nimbleway.com/web

NIMBLE_API_KEY = os.environ.get("NIMBLE_API_KEY")
KAFKA_BOOTSTRAP = os.environ.get("KAFKA_BOOTSTRAP", "localhost:9092")
TOPIC = os.environ.get("TOPIC", "nimble.pricing")

producer = KafkaProducer(
 bootstrap_servers=KAFKA_BOOTSTRAP,
 value_serializer=lambda v: json.dumps(v).encode("utf-8"),
)

headers = {"Authorization": f"Bearer {NIMBLE_API_KEY}", "Content-Type": "application/json"}

# Example Nimble request body: fetch a product page & return structured JSON

def fetch_nimble_event(url: str) -> dict:
 body = {
 "url": url,
 "parse": True,
 "schema": {
 "_meta": {"source":"string","country":"string","collected_at":"string"},
 "product": {"id":"string","title":"string","brand":"string"},
 "offer": {"price":"number","currency":"string","in_stock":"boolean","seller":"string","delivery":"string"},
 "serp": {"keyword":"string","rank":"number"},
 "page": {"url":"string"}
 }
 }
 r = requests.post(NIMBLE_API_URL, headers=headers, json=body, timeout=30)
 r.raise_for_status()
 return r.json()

urls = [
 "(product URL)",
 "(product URL)",
]

while True:
 for u in urls:
 try:
 evt = fetch_nimble_event(u)

# attach standard metadata if needed

 evt.setdefault("_meta", {})
 evt["_meta"].setdefault("collected_at", time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))
 producer.send(TOPIC, evt)
 except Exception as e:
 print("warn:", e)
 producer.flush()
 time.sleep(15)

4) Validate in Snowflake

SELECT COUNT(*) FROM NIMBLE_DEMO.RETAIL.PRICING_BRONZE; -- rows arriving?
SELECT * FROM NIMBLE_DEMO.RETAIL.PRICING_SILVER LIMIT 20; -- typed projection
SELECT * FROM NIMBLE_DEMO.RETAIL.PRICING_GOLD LIMIT 20; -- business view

Part B — Databricks Auto Loader + Delta Live Tables (Medallion)

Nimble can land JSON to S3/GCS in real time, which Auto Loader ingests with schema evolution. DLT then materializes Silver/Gold.

1) AWS IAM for S3 access (Databricks)

Trust policy (role assumed by Databricks; fill in your Databricks account/external IDs):

{
 "Version": "2012-10-17",
 "Statement": [{
 "Effect": "Allow",
 "Principal": {"AWS": "arn:aws:iam::<DATABRICKS_ACCOUNT_ID>:root"},
 "Action": "sts:AssumeRole",
 "Condition": {"StringEquals": {"sts:ExternalId": "<DATABRICKS_EXTERNAL_ID>"}}
 }]
}

Permissions policy (restrict to your buckets/prefixes):

{
 "Version": "2012-10-17",
 "Statement": [
 {"Effect": "Allow","Action": ["s3:ListBucket"],"Resource": ["arn:aws:s3:::nimble-data-prod"],
 "Condition": {"StringLike": {"s3:prefix": ["bronze/pricing/*","_checkpoint/*"]}}},
 {"Effect": "Allow","Action": ["s3:GetObject"],"Resource": ["arn:aws:s3:::nimble-data-prod/bronze/pricing/*"]},
 {"Effect": "Allow","Action": ["s3:PutObject","s3:DeleteObject"],"Resource": ["arn:aws:s3:::nimble-data-prod/_checkpoint/*"]}
 ]
}

Attach this role as your Databricks instance profile used by Auto Loader/DLT.

2) DLT pipeline (Python) — Bronze/Silver/Gold

Create a DLT pipeline with this notebook as the source. Set the S3 path and checkpoint location.

import dlt
from pyspark.sql.functions import col, to_timestamp

BRONZE_PATH = "s3://nimble-data-prod/bronze/pricing/"

# Nimble landing path

@dlt.table(name="pricing_bronze", comment="Raw Nimble pricing events (JSON)")
def pricing_bronze():
 return (
 spark.readStream.format("cloudFiles").option("cloudFiles.format", "json").option("cloudFiles.inferColumnTypes", "true").option("cloudFiles.schemaLocation", "s3://nimble-data-prod/_checkpoint/pricing_schema/").load(BRONZE_PATH)
 )

@dlt.table(name="pricing_silver", comment="Parsed & typed fields from Nimble events")
def pricing_silver():
 df = dlt.read_stream("pricing_bronze")
 return (
 df.select(
 col("_meta.source").alias("source"),
 col("_meta.country").alias("country"),
 to_timestamp(col("_meta.collected_at")).alias("collected_at"),
 col("product.id").alias("product_id"),
 col("product.title").alias("title"),
 col("offer.price").cast("double").alias("price"),
 col("offer.currency").alias("currency"),
 col("offer.in_stock").cast("boolean").alias("in_stock"),
 col("offer.seller").alias("seller"),
 col("serp.rank").cast("int").alias("serp_rank"),
 col("page.url").alias("url")
 )
 )

@dlt.table(name="pricing_gold", comment="Current price per product/source with last seen time")
def pricing_gold():
 from pyspark.sql.window import Window
 from pyspark.sql.functions import row_number

 df = dlt.read_stream("pricing_silver")
 w = Window.partitionBy("product_id","source").orderBy(col("collected_at").desc())
 latest = (df.withColumn("rn", row_number().over(w)).filter(col("rn") == 1).drop("rn"))
 return latest

Run the pipeline; DLT materializes Delta tables for each layer with continuous updates.

3) Query examples (Databricks SQL)

SELECT * FROM LIVE.pricing_bronze LIMIT 5;
SELECT * FROM LIVE.pricing_silver WHERE in_stock = TRUE ORDER BY collected_at DESC LIMIT 20;
SELECT product_id, source, price, currency, collected_at FROM LIVE.pricing_gold ORDER BY collected_at DESC LIMIT 20;

Operational guidance

Data quality & governance: Nimble enforces schema, lineage, and PII masking in its platform layer; use these outputs directly for Silver/Gold.
Cost/perf tips (Auto Loader): tune cloudFiles.inferColumnTypes, compact small files (OPTIMIZE/ZORDER), and set schema hints for hot fields.
Kafka/Snowpipe Streaming: use batching (buffer.count.records, buffer.flush.time) and partition topics by domain/use case (e.g., nimble.pricing, nimble.serp).
Compliance & logging: align with Nimble’s compliance‑by‑design and observability.

Variations (when streaming isn’t available)

File Snowpipe (auto‑ingest): land JSON to S3 via Nimble, configure Snowflake stage + pipe to load into Bronze, then reuse the Silver/Gold SQL above. Nimble natively streams to S3/GCS.
Direct to warehouse: Nimble can stream governed tables directly to Snowflake/Databricks when preferred.

Checklist

[ ] Kafka connector deployed with snowflake.ingestion.method=SNOWPIPE_STREAMING and writing to PRICING_BRONZE.
[ ] Bronze/Silver/Gold objects created and returning rows.
[ ] Auto Loader reading from Nimble landing bucket; DLT producing Silver/Gold.
[ ] IAM role attached to Databricks with least‑privilege S3 access.
[ ] Alerts/monitoring set in Nimble Analytics & Management.

Why Nimble for Medallion pipelines

Nimble’s agents browse live pages (not cached indices), normalize and validate outputs, and stream them into Snowflake/Databricks with zero maintenance, yielding higher data freshness and lower breakage across industry use cases.