Modernizing NoSQL Data: A Cassandra to Snowflake Integration Blueprint

Moving data from Apache Cassandra to Snowflake using Snowpark Python allows for a highly efficient pipeline that leverages native Python drivers while maintaining Snowflake’s governed security and scale. This approach is particularly effective for high-scale source systems where maintaining full change history (SCD Type 2) is a priority.

Apache Cassandra (The Source)

Apache Cassandra is a distributed, wide-column NoSQL database designed to handle large volumes of data across many commodity servers.

• Data Model: It uses a flexible, wide-column schema that is ideal for structured data that needs to scale horizontally.
• Use Case: It is frequently used for high-velocity applications such as product catalogs, where attributes like price and stock levels change frequently and need to be captured reliably.
• Connectivity: For integration purposes, the official DataStax cassandra-driver allows Python applications to read rows directly as native Python objects (tuples/lists), avoiding the need for intermediate file formats.

Snowpark (The Engine)

Snowpark is the set of libraries and runtimes in Snowflake that allow developers to use non-SQL languages, such as Python, to build data pipelines directly within the Snowflake environment.

• Python Native: Because a Snowpark session is essentially a standard Python environment, the entire Python ecosystem — including specialized database drivers like cassandra-driver — is available for use.
• DataFrame API: Snowpark provides a DataFrame API that translates complex Python logic into a single SQL execution plan, which Snowflake then executes on its high-performance engine.
• Serverless Execution: Logic can be productionized inside Snowflake using native schedulers like Snowflake Tasks or hosted as containerized jobs through Snowpark Container Services (SPCS).
• Efficiency: By declaring explicit schemas (using StructType), developers can bypass the need for pandas or pyarrow in the data path, reducing memory overhead and avoiding common type-inference errors.

High-Level Architecture

The transition from Cassandra to Snowflake via Snowpark typically follows this structured flow:

The Use Case

The source system is a product catalogue living in Cassandra: ten thousand-ish SKUs in the real thing, ten rows in the demo keyspace. Prices change. Stock levels change. Occasionally a product gets renamed or moved between categories. Analysts in Snowflake need two things from this, and they conflict:

• The current truth — one row per product, latest values, fast to query for dashboards.
• The full history — every version of every product, with the window of time it was valid, so margin and pricing analysis can be run as of any date.A plain overwrite gives you the first and destroys the second. An append-only load gives you a pile of duplicates with no way to tell which row is live. The answer is a Slowly Changing Dimension Type2 table: when a tracked attribute changes, expire the old row by stamping EFFECTIVE_TO and flipping IS_CURRENT to false, then insert the new version alongside it.

Four columns are treated as history-worthy: NAME, CATEGORY, PRICE and STOCK_QTY. A change in any of them creates a new version; a change in anything else does not.

Tech Stack

• Apache Cassandra 4.x is the source, queried with the DataStax cassandra-driver.
• Python 3.11 runs the pipeline itself.
• Snowpark Python provides typed DataFrames and the write path.
• Snowflake provides storage: a curated snapshot table plus an SCD2 history table.
• Orchestration is Snowflake-native: either a Python stored procedure driven by a Task, or a container job on Snowpark Container Services.
• Docker and docker-compose run Cassandra locally for the demo.
• uv handles dependency management.

Snowflake Setup

-- Snowflake DDL for Cassandra → Snowflake pipeline
-- Database: SJAYABALDB  Schema: COCO
USE DATABASE SJAYABALDB;
USE SCHEMA COCO;

-- Note: there is no separate raw landing table. Snowpark receives typed tuples
-- straight from the Cassandra driver via an explicit StructType, so the extract
-- lands directly in the curated and SCD2 tables below.

-- SCD Type 2 history table
-- Natural key : ID  (UUID from Cassandra)
-- Tracked cols: NAME, CATEGORY, PRICE, STOCK_QTY
-- Each change creates a new row; old row is expired (EFFECTIVE_TO set, IS_CURRENT = FALSE)
CREATE TABLE IF NOT EXISTS CASSANDRA_PRODUCTS_SCD2 (
   -- Surrogate key (one per version)
   SCD_KEY          VARCHAR(36)    NOT NULL,
   -- Natural / business key
   ID               VARCHAR(36)    NOT NULL,
   -- Source columns
   NAME             VARCHAR(255),
   CATEGORY         VARCHAR(100),
   PRICE            NUMBER(10, 2),
   STOCK_QTY        INTEGER,
   CREATED_AT       TIMESTAMP_NTZ,
   -- Derived / transformed columns (same as curated)
   PRICE_TIER       VARCHAR(20),
   STOCK_VALUE      NUMBER(14, 2),
   IS_HIGH_VALUE    BOOLEAN,
   SOURCE_SYSTEM    VARCHAR(50),
   PIPELINE_RUN_ID  VARCHAR(36),
   -- SCD2 tracking columns
   EFFECTIVE_FROM   TIMESTAMP_NTZ  NOT NULL,
   EFFECTIVE_TO     TIMESTAMP_NTZ,               -- NULL  = currently active row
   IS_CURRENT       BOOLEAN        NOT NULL DEFAULT TRUE
);

-- Curated table: raw + Snowpark-derived transformations
CREATE TABLE IF NOT EXISTS CASSANDRA_PRODUCTS_CURATED (
   -- Source columns
   ID               VARCHAR(36)    NOT NULL,
   NAME             VARCHAR(255),
   CATEGORY         VARCHAR(100),
   PRICE            NUMBER(10, 2),
   STOCK_QTY        INTEGER,
   CREATED_AT       TIMESTAMP_NTZ,
   -- Derived / transformed columns
   PRICE_TIER       VARCHAR(20),        -- Budget / Mid-Range / Premium
   STOCK_VALUE      NUMBER(14, 2),      -- PRICE * STOCK_QTY
   IS_HIGH_VALUE    BOOLEAN,            -- STOCK_VALUE > 10000
   SOURCE_SYSTEM    VARCHAR(50),        -- always 'CASSANDRA'
   -- Pipeline audit columns
   PIPELINE_RUN_ID  VARCHAR(36),        -- UUID per pipeline run
   LOADED_AT        TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

%pip install cassandra-driver
import os
from cassandra.cluster import Cluster
from snowflake.snowpark import Session
from snowflake.snowpark.types import StructType, StructField, StringType, IntegerType, DecimalType

# ── Cassandra config (env vars override hardcoded defaults) ───────────────────
CASSANDRA_HOST  = os.environ.get("CASSANDRA_HOST",  "127.0.0.1")
CASSANDRA_PORT  = int(os.environ.get("CASSANDRA_PORT",  "9042"))
KEYSPACE        = os.environ.get("CASSANDRA_KEYSPACE", "demo_ks")
CASSANDRA_TABLE = os.environ.get("CASSANDRA_TABLE", "products")

# ── Snowflake config (env vars override hardcoded defaults) ───────────────────
SF_CONNECTION    = os.environ.get("SF_CONNECTION",    "cassandra_pipeline")
SF_DATABASE      = os.environ.get("SF_DATABASE",      "SJAYABALDB")
SF_SCHEMA        = os.environ.get("SF_SCHEMA",        "COCO")
SF_TABLE_CURATED = os.environ.get("SF_TABLE_CURATED", "CASSANDRA_PRODUCTS_CURATED")
SF_TABLE_SCD2    = os.environ.get("SF_TABLE_SCD2",    "CASSANDRA_PRODUCTS_SCD2")
SF_WAREHOUSE     = os.environ.get("SF_WAREHOUSE",     "COMPUTE_WH")

def get_session():
    return Session.builder.config("connection_name", SF_CONNECTION).create()

def extract_cassandra_data():
    """Fetches data from Cassandra as native Python objects."""
    cluster = Cluster([CASSANDRA_HOST], port=CASSANDRA_PORT)
    session = cluster.connect(KEYSPACE)
    query = f"SELECT * FROM {CASSANDRA_TABLE}"
    rows = session.execute(query)
    # Convert result set to list of tuples for Snowpark DataFrame creation
    return [tuple(row) for row in rows]

# Example usage pattern within your pipeline:
# session = get_session()
# raw_data = extract_cassandra_data()
# 
# df = session.create_dataframe(raw_data, schema=...)
# df.write.mode("overwrite").save_as_table(
#     f"{SF_DATABASE}.{SF_SCHEMA}.{SF_TABLE_CURATED}")

Code Snippets

This is the part worth reading closely. The pipeline is a single module with four moving parts: read,type, transform, and version.

1 · Read from Cassandra into plain tuples

The driver hands back row objects with real Python types — uuid.UUID, Decimal, datetime. I normalise the UUID to a string and strip the timezone so the value lands cleanly in TIMESTAMP_NTZ,then return a list of tuples in schema order. No DataFrame yet.To complete the “Read” phase of your pipeline, you can use the following function to extract data from Cassandra, normalise the types (e.g., UUID to string, stripping timezones for TIMESTAMP_NTZ), and return a list of tuples ready for Snowpark DataFrame creation.

from cassandra.cluster import Cluster

def read_from_cassandra() -> list[tuple]:
    """Reads products from Cassandra and returns as list of tuples."""
    # No protocol_version pinned — the driver negotiates with the cluster.
    # Pin it only if you must (protocol 5 requires Cassandra 4.0+; 3.x
    # clusters generally need 3 or 4).
    cluster = Cluster(contact_points=[CASSANDRA_HOST], port=CASSANDRA_PORT)
    try:
        session = cluster.connect(KEYSPACE)
        rows = session.execute(
            "SELECT id, name, category, price, stock_qty, created_at FROM products"
        )
        # Normalise types for Snowpark compatibility
        return [
            (
                str(row.id),
                row.name,
                row.category,
                row.price,
                row.stock_qty,
                row.created_at.replace(tzinfo=None) if row.created_at else None,
            )
            for row in rows
        ]
    finally:
        cluster.shutdown()

2 · Declare the schema, skip pandas entirely

An explicit StructType is the whole trick. Snowpark gets exact types up front, so no inference, no pyarrow round-trip, and no pandas dependency in the data path.

from snowflake.snowpark.types import StructType, StructField, StringType, IntegerType, DecimalType, TimestampType

# Explicit schema declaration avoids type inference and pandas dependencies
RAW_SCHEMA = StructType([
    StructField("ID", StringType()),
    StructField("NAME", StringType()),
    StructField("CATEGORY", StringType()),
    StructField("PRICE", DecimalType(10, 2)),
    StructField("STOCK_QTY", IntegerType()),
    StructField("CREATED_AT", TimestampType()),
])

# Create the Snowpark DataFrame directly from the extracted records
raw_df = session.create_dataframe(records, schema=RAW_SCHEMA)

3 · Derive business columns with the DataFrame API

Five derived columns, all expressed as Snowpark expressions. Nothing is computed in Python — this compiles down to a single SQL plan that Snowflake executes.

from snowflake.snowpark.functions import col, lit, when
from snowflake.snowpark.types import DecimalType

def apply_transformations(df, pipeline_run_id: str):
    """
    Applies business logic to the raw product DataFrame using 
    Snowpark expressions.
    """
    return (
        df.with_column(
            "PRICE_TIER",
            when(col("PRICE") < 50, lit("Budget"))
            .when(col("PRICE") <= 500, lit("Mid-Range"))
            .otherwise(lit("Premium")),
        )
        .with_column(
            "STOCK_VALUE", 
            (col("PRICE") * col("STOCK_QTY")).cast(DecimalType(14, 2))
        )
        .with_column("IS_HIGH_VALUE", col("STOCK_VALUE") > lit(10000))
        .with_column("SOURCE_SYSTEM", lit("CASSANDRA"))
        .with_column("PIPELINE_RUN_ID", lit(pipeline_run_id))
    )

4 · Load the curated table

import uuid
from snowflake.snowpark.functions import current_timestamp

# run_id is created once in main() and passed to *both* loads, so the
# curated snapshot and the SCD2 versions share one audit identifier.
def load_curated(session, records, run_id: str) -> int:
    raw_df = session.create_dataframe(records, schema=RAW_SCHEMA)

    # Apply transformations and add audit metadata column
    transformed_df = apply_transformations(raw_df, run_id).with_column(
        "LOADED_AT", current_timestamp()
    )

    # overwrite, not append: this table is the current-truth snapshot.
    # Appending would re-add the whole extract on every run.
    transformed_df.write.mode("overwrite").save_as_table(
        f"{SF_DATABASE}.{SF_SCHEMA}.CASSANDRA_PRODUCTS_CURATED"
    )
    return len(records)

5 · Detect change in Python, not in a join

Here I deliberately stepped away from the DataFrame API. DataFrame.alias() is marked experimental in Snowpark 1.54, and in a self-join it mangles column names — ID comes back as IDSRC and the next collect() dies on an invalid identifier. One collect() of the active rows into a dict is simpler, and at this data volume it is also faster.

# 5 · Detect change in Python, not in a join
# Fetch current active records from Snowflake
current_rows = (
    session.table(fqn)
    .filter(col("IS_CURRENT") == lit(True))
    .select("ID", *SCD2_TRACKED_COLS, "EFFECTIVE_FROM")
    .collect()
)

# Index by ID for O(1) lookup
current_index = {r["ID"]: {c: r[c] for c in SCD2_TRACKED_COLS} for r in current_rows}

def _normalize(value):
    # Coerce numeric values to Decimal before comparing. The Cassandra driver
    # returns float for PRICE, collect() returns Decimal, and
    # Decimal('29.99') != 29.99 in Python, so mixed types compare unequal.
    if isinstance(value, float):
        return Decimal(str(value))
    return value

# Compare source rows against indexed state
new_rows = []
changed_rows = []
expired_ids = []
unchanged_ids = []

for row in src_rows:
    rid = row["ID"]
    if rid not in current_index:
        new_rows.append(row)  # never seen -> insert
    elif any(_normalize(row[c]) != _normalize(current_index[rid][c]) for c in SCD2_TRACKED_COLS):
        changed_rows.append(row)  # tracked value moved
        expired_ids.append(rid)   # flag old version to expire
    else:
        unchanged_ids.append(rid) # no-op

Watch the types on both sides of that comparison. PRICE comes back as a Decimal when collect() reads a Snowflake table, but the Cassandra driver hands back a plain float. In Python, Decimal(‘29.99’) != 29.99 evaluates to True, so a bare row[c] != current_index[rid][c] check would flag every row as changed on every run and generate phantom SCD2 versions. The _normalize() helper coerces floats to Decimal before comparing, so PRICE (and any other numeric tracked column) is compared like-for-like.

6 · Expire the old versions with a MERGE

The IDs to retire go into a session-scoped temporary table, then a single MERGE stamps them closed. The temp table is created before the transaction opens: DDL issued mid-transaction triggers an implicit commit in Snowflake, which would quietly defeat the atomicity set up here. Using CREATE TEMPORARY (rather than a regular table named after the run) also removes any chance of two overlapping runs clobbering each other, and it disappears with the session.

# 6 · Expire the old versions with a MERGE
from snowflake.snowpark.types import StructType, StructField, StringType
#Creating named temp objects is not supported in an owner's rights stored procedure.CREATE OR REPLACE PROCEDURE defaults to EXECUTE AS OWNER, so the SCD2 step fails when the pipeline runs via the Task → stored proc path.Hence execute as CALLER instead OWNER
fqn = f"{SF_DATABASE}.{SF_SCHEMA}.{SF_TABLE_SCD2}"
tmp = f"TMP_SCD2_EXPIRE_{run_id.replace('-', '_')}"

# Session-scoped temp table, created BEFORE the transaction (DDL inside a
# transaction causes an implicit commit and breaks atomicity).
expire_schema = StructType([StructField("ID", StringType())])
session.create_dataframe(
    [(eid,) for eid in expired_ids], schema=expire_schema
).write.mode("overwrite").save_as_table(tmp, table_type="temporary")

# Expire + insert must land together, so both run in one transaction.
session.sql("BEGIN").collect()

session.sql(f"""
    MERGE INTO {fqn} AS tgt
    USING {tmp} AS exp ON tgt.ID = exp.ID AND tgt.IS_CURRENT = TRUE
    WHEN MATCHED THEN UPDATE SET
        tgt.EFFECTIVE_TO = CURRENT_TIMESTAMP(),
        tgt.IS_CURRENT   = FALSE
""").collect()

7 · Insert the new versions

New products and new versions of changed products are inserted together, each with a fresh surrogate key, an open-ended validity window, and IS_CURRENT = True. This runs inside the transaction opened in step 6 and commits at the end, so the expire and the insert either both land or neither does. Without that, a crash between the two leaves rows expired with no live successor — every version of a product closed and nothing marked current.

# 7 · Insert the new versions (same transaction as step 6)
from datetime import datetime, timezone
import uuid
from snowflake.snowpark.types import (
    StructType, StructField, StringType, IntegerType,
    DecimalType, TimestampType, BooleanType,
)

scd2_schema = StructType([
    StructField("SCD_KEY", StringType()),
    StructField("ID", StringType()),
    StructField("NAME", StringType()),
    StructField("CATEGORY", StringType()),
    StructField("PRICE", DecimalType(10, 2)),
    StructField("STOCK_QTY", IntegerType()),
    StructField("CREATED_AT", TimestampType()),
    StructField("PRICE_TIER", StringType()),
    StructField("STOCK_VALUE", DecimalType(14, 2)),
    StructField("IS_HIGH_VALUE", BooleanType()),
    StructField("SOURCE_SYSTEM", StringType()),
    StructField("PIPELINE_RUN_ID", StringType()),
    StructField("EFFECTIVE_FROM", TimestampType()),
    StructField("EFFECTIVE_TO", TimestampType()),
    StructField("IS_CURRENT", BooleanType()),
])

effective_now = datetime.now(timezone.utc).replace(tzinfo=None)

scd2_records = [
    (
        str(uuid.uuid4()), # SCD_KEY — one per version
        row["ID"], row["NAME"], row["CATEGORY"], row["PRICE"], row["STOCK_QTY"],
        row["CREATED_AT"], row["PRICE_TIER"], row["STOCK_VALUE"],
        row["IS_HIGH_VALUE"], row["SOURCE_SYSTEM"], row["PIPELINE_RUN_ID"],
        effective_now, None, True, # EFFECTIVE_FROM, EFFECTIVE_TO (NULL=live), IS_CURRENT
    )
    for row in (new_rows + changed_rows)
]

session.create_dataframe(scd2_records, schema=scd2_schema) \
    .write.mode("append").save_as_table(fqn)

# Commit both steps; roll back everything if either failed.
session.sql("COMMIT").collect()
# except Exception:
#     session.sql("ROLLBACK").collect()
#     raise                                                               

-- Current truth: drop-in replacement for a plain table
SELECT * 
FROM SJAYABALDB.COCO.CASSANDRA_PRODUCTS_SCD2
WHERE IS_CURRENT = TRUE;

-- Price history for one product
SELECT NAME, PRICE, EFFECTIVE_FROM, EFFECTIVE_TO, IS_CURRENT
FROM SJAYABALDB.COCO.CASSANDRA_PRODUCTS_SCD2
WHERE ID = '56dc723a-e10d-4deb-8db6-0d1fa77e519f'
ORDER BY EFFECTIVE_FROM;

Productionize

Config leaves the source tree

Hosts, keyspace, warehouse, and table names default to the demo values and override from env.Secrets never live in git. Locally you copy .env.example. The stored-procedure path takes Cassandra host as arguments. The SPCS path mounts an RSA key as a Snowflake secret inside the container.

.env.example
CASSANDRA_HOST=cassandra.prod.internal
CASSANDRA_PORT=9042
CASSANDRA_KEYSPACE=demo_ks
SF_DATABASE=SJAYABALDB
SF_SCHEMA=COCO
SF_WAREHOUSE=COMPUTE_WH
SF_CONNECTION=cassandra_pipeline
# Used by the SPCS container, which has no connections.toml
SNOWFLAKE_ACCOUNT=...
SNOWFLAKE_USER=...
SNOWFLAKE_PRIVATE_KEY_PATH=/secrets/rsa_key.p8

def get_snowflake_session() -> Session:
if _INJECTED_SESSION is not None:
return _INJECTED_SESSION # stored proc
account = os.environ.get("SNOWFLAKE_ACCOUNT")
user = os.environ.get("SNOWFLAKE_USER")
if account and user: # SPCS container
builder = (Session.builder
.config("account", account).config("user", user)
.config("database", SF_DATABASE).config("warehouse", SF_WAREHOUSE))
if os.environ.get("SNOWFLAKE_PRIVATE_KEY_PATH"):
builder = builder.config("private_key_file",
os.environ["SNOWFLAKE_PRIVATE_KEY_PATH"])
return builder.create()
return Session.builder.config("connection_name", SF_CONNECTION).create() # local

Both paths need egress to Cassandra

Snowflake cannot reach Cassandra until you say so. A network rule names the host and port; an External Access Integration (ACCOUNTADMIN) attaches that rule. The stored procedure and the SPCS job both declare EXTERNAL_ACCESS_INTEGRATIONS = (CASSANDRA_EAI).

CREATE OR REPLACE NETWORK RULE CASSANDRA_EGRESS_RULE
TYPE = HOST_PORT MODE = EGRESS
VALUE_LIST = ('cassandra.prod.internal:9042');
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION CASSANDRA_EAI
ALLOWED_NETWORK_RULES = (SJAYABALDB.COCO.CASSANDRA_EGRESS_RULE)
ENABLED = TRUE;

Path 1 — Snowflake orchestration

The worker is a Python stored procedure. The scheduler is a Task that CALLs it every hour. Code is a zip on an internal stage — pipeline.py, stored_proc_handler.py, and cassandra-driver. Snowpark is provided by the warehouse; do not bundle it. The handler injects Snowflake’s session so the pipeline never opens a second one.

One packaging detail that will bite you: cassandra-driver ships compiled C extensions, and the warehouse runs Linux x86_64. If you build the zip on a Mac you will get macOS (or arm64) artifacts and the stored procedure fails at import time, not at deploy time. Resolve the wheels for the target platform instead of the local one — with uv that means uv pip install — target ./pkg — python-version 3.11 — python-platform x86_64-manylinux2014 — only-binary=:all: — no-deps cassandra-driver, and with plain pip the equivalent is pip install — platform manylinux2014_x86_64 — only-binary=:all: — target ./pkg cassandra-driver. Forcing binary-only resolution is the important part: it stops pip from quietly building the extension for whatever machine you happen to be on. The same applies to the SPCS path, where the fix is building the image with — platform linux/amd64.Please install %pip install cassandra-driver

"""
pipeline.py
Reads all rows from Cassandra demo_ks.products and:
  1. Loads curated data into CASSANDRA_PRODUCTS_CURATED
  2. Applies SCD Type 2 logic into CASSANDRA_PRODUCTS_SCD2

SCD2 logic (pure Snowpark):
  • New IDs     → INSERT as IS_CURRENT=TRUE
  • Changed IDs → EXPIRE old row (EFFECTIVE_TO, IS_CURRENT=FALSE)
                  INSERT new version as IS_CURRENT=TRUE
  • Unchanged   → skip

No pandas anywhere — Python tuples → session.create_dataframe(schema).

Usage:
    uv run src/pipeline.py
"""

import logging
import os
import uuid
from datetime import datetime, timezone
from decimal import Decimal

from cassandra.cluster import Cluster
from snowflake.snowpark import Session
from snowflake.snowpark.types import (
    StructType, StructField,
    StringType, DecimalType, IntegerType, TimestampType, BooleanType,
)
from snowflake.snowpark.functions import (
    col, lit, when, current_timestamp,
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
log = logging.getLogger(__name__)

# ── Cassandra config (env vars override hardcoded defaults) ───────────────────
CASSANDRA_HOST  = os.environ.get("CASSANDRA_HOST",  "127.0.0.1")
CASSANDRA_PORT  = int(os.environ.get("CASSANDRA_PORT",  "9042"))
KEYSPACE        = os.environ.get("CASSANDRA_KEYSPACE", "demo_ks")
CASSANDRA_TABLE = os.environ.get("CASSANDRA_TABLE", "products")

# ── Snowflake config (env vars override hardcoded defaults) ───────────────────
SF_CONNECTION    = os.environ.get("SF_CONNECTION",    "spark-connect")
SF_DATABASE      = os.environ.get("SF_DATABASE",      "SJAYABALDB")
SF_SCHEMA        = os.environ.get("SF_SCHEMA",        "COCO")
SF_TABLE_CURATED = os.environ.get("SF_TABLE_CURATED", "CASSANDRA_PRODUCTS_CURATED")
SF_TABLE_SCD2    = os.environ.get("SF_TABLE_SCD2",    "CASSANDRA_PRODUCTS_SCD2")
SF_WAREHOUSE     = os.environ.get("SF_WAREHOUSE",     "COMPUTE_WH")

# Set by stored_proc_handler when running inside Snowflake (avoids creating a
# second session — Snowflake injects one for us).
_INJECTED_SESSION: Session | None = None

# Columns that trigger a new SCD2 version when their value changes
SCD2_TRACKED_COLS = ["NAME", "CATEGORY", "PRICE", "STOCK_QTY"]

# ── Snowpark schema matching Cassandra's products table ───────────────────────
RAW_SCHEMA = StructType([
    StructField("ID",         StringType()),
    StructField("NAME",       StringType()),
    StructField("CATEGORY",   StringType()),
    StructField("PRICE",      DecimalType(10, 2)),
    StructField("STOCK_QTY",  IntegerType()),
    StructField("CREATED_AT", TimestampType()),
])


# ── Cassandra helpers ─────────────────────────────────────────────────────────

def read_from_cassandra() -> list[tuple]:
    """
    Connect to local Docker Cassandra and return rows as a list of tuples
    matching RAW_SCHEMA column order. Python datetime objects are passed
    directly — no pandas conversion needed.
    """
    log.info("Connecting to Cassandra %s:%s ...", CASSANDRA_HOST, CASSANDRA_PORT)
    from cassandra.policies import DCAwareRoundRobinPolicy
    from cassandra.io.asyncioreactor import AsyncioConnection
    # protocol_version=5 requires Cassandra 4.0+. Leave unset (None) to let the
    # driver auto-negotiate against the cluster, or pin it via env var if you
    # need a specific version (e.g. 3.x clusters typically need protocol 3 or 4).
    protocol_version_env = os.environ.get("CASSANDRA_PROTOCOL_VERSION")
    cluster = Cluster(
        contact_points=[CASSANDRA_HOST],
        port=CASSANDRA_PORT,
        protocol_version=int(protocol_version_env) if protocol_version_env else None,
        load_balancing_policy=DCAwareRoundRobinPolicy(local_dc="datacenter1"),
        connection_class=AsyncioConnection,
    )
    try:
        session = cluster.connect(KEYSPACE)
        log.info("Fetching rows from %s.%s ...", KEYSPACE, CASSANDRA_TABLE)
        rows = session.execute(
            f"SELECT id, name, category, price, stock_qty, created_at FROM {CASSANDRA_TABLE}"
        )
        records = [
            (
                str(row.id),
                row.name,
                row.category,
                row.price,
                row.stock_qty,
                row.created_at.replace(tzinfo=None) if row.created_at else None,
            )
            for row in rows
        ]
        log.info("Read %d rows from Cassandra.", len(records))
        return records
    finally:
        cluster.shutdown()


# ── Snowflake / Snowpark helpers ──────────────────────────────────────────────

def get_snowflake_session() -> Session:
    if _INJECTED_SESSION is not None:
        log.info("Using Snowflake session injected by stored proc handler.")
        return _INJECTED_SESSION

    # If explicit account/user env vars are set (e.g. Docker), build the session
    # directly without relying on a named connection profile.
    account = os.environ.get("SNOWFLAKE_ACCOUNT")
    user    = os.environ.get("SNOWFLAKE_USER")
    if account and user:
        log.info("Connecting to Snowflake via env-var credentials (account=%s) ...", account)
        builder = (
            Session.builder
            .config("account",  account)
            .config("user",     user)
            .config("role",     os.environ.get("SNOWFLAKE_ROLE", ""))
            .config("database", SF_DATABASE)
            .config("schema",   SF_SCHEMA)
            .config("warehouse", SF_WAREHOUSE)
        )
        private_key_path = os.environ.get("SNOWFLAKE_PRIVATE_KEY_PATH")
        if private_key_path:
            builder = builder.config("private_key_file", private_key_path)
            passphrase = os.environ.get("SNOWFLAKE_PRIVATE_KEY_PASSPHRASE", "")
            if passphrase:
                builder = builder.config("private_key_file_pwd", passphrase)
        else:
            builder = builder.config("password", os.environ.get("SNOWFLAKE_PASSWORD", ""))
        return builder.create()

    log.info("Connecting to Snowflake (connection: %s) ...", SF_CONNECTION)
    return Session.builder.config("connection_name", SF_CONNECTION).create()


def apply_transformations(df, pipeline_run_id: str):
    """
    Add derived columns using pure Snowpark DataFrame API.

    PRICE_TIER      – Budget / Mid-Range / Premium
    STOCK_VALUE     – PRICE × STOCK_QTY
    IS_HIGH_VALUE   – STOCK_VALUE > 10,000
    SOURCE_SYSTEM   – literal 'CASSANDRA'
    PIPELINE_RUN_ID – UUID per pipeline run
    """
    return (
        df
        .with_column(
            "PRICE_TIER",
            when(col("PRICE") < 50,    lit("Budget"))
            .when(col("PRICE") <= 500, lit("Mid-Range"))
            .otherwise(                lit("Premium")),
        )
        .with_column("STOCK_VALUE",    (col("PRICE") * col("STOCK_QTY")).cast(DecimalType(14, 2)))
        .with_column("IS_HIGH_VALUE",  col("STOCK_VALUE") > lit(10000))
        .with_column("SOURCE_SYSTEM",  lit("CASSANDRA"))
        .with_column("PIPELINE_RUN_ID", lit(pipeline_run_id))
    )


# ── Curated load ──────────────────────────────────────────────────────────────

def load_curated(session: Session, records: list[tuple], run_id: str) -> int:
    """
    Overwrites CASSANDRA_PRODUCTS_CURATED with the latest full extract each run.

    This table is a point-in-time snapshot ("current truth for dashboards"),
    not an append-only log — every run replaces its contents. History lives in
    CASSANDRA_PRODUCTS_SCD2, not here.
    """
    if not records:
        log.warning("No rows to load.")
        return 0

    raw_df = session.create_dataframe(records, schema=RAW_SCHEMA)
    transformed_df = apply_transformations(raw_df, run_id).with_column(
        "LOADED_AT", current_timestamp()
    )

    log.info("Writing %d rows to %s (run_id=%s) ...", len(records), SF_TABLE_CURATED, run_id)
    transformed_df.write.mode("overwrite").save_as_table(
        f"{SF_DATABASE}.{SF_SCHEMA}.{SF_TABLE_CURATED}"
    )
    log.info("Curated load complete.")
    return len(records)


# ── SCD Type 2 ────────────────────────────────────────────────────────────────

def apply_scd2(session: Session, records: list[tuple], run_id: str) -> dict:
    """
    SCD Type 2 upsert into CASSANDRA_PRODUCTS_SCD2.

    Approach: collect current SCD2 state into Python dicts, compare locally,
    then push only the necessary inserts/updates back to Snowflake.
    This avoids Snowpark join aliasing issues and keeps the logic explicit.

    Returns a dict with counts: new, expired, inserted, unchanged.
    """
    if not records:
        log.warning("No rows — skipping SCD2.")
        return {}

    fqn = f"{SF_DATABASE}.{SF_SCHEMA}.{SF_TABLE_SCD2}"

    # ── Pull current active rows from SCD2 table ───────────────────────────
    current_rows = (
        session.table(fqn)
        .filter(col("IS_CURRENT") == lit(True))
        .select("ID", *SCD2_TRACKED_COLS, "EFFECTIVE_FROM")
        .collect()
    )
    # Keyed by ID → {col: value, ...}
    current_index = {
        r["ID"]: {c: r[c] for c in SCD2_TRACKED_COLS + ["EFFECTIVE_FROM"]}
        for r in current_rows
    }

    # ── Classify incoming records ───────────────────────────────────────────
    # Build transformation on the full source first to get derived columns
    raw_df = session.create_dataframe(records, schema=RAW_SCHEMA)
    src_df = apply_transformations(raw_df, run_id)
    src_rows = src_df.collect()

    new_rows      = []   # IDs never seen before
    changed_rows  = []   # IDs that exist but tracked values changed
    expired_ids   = []   # IDs to mark as expired in target
    unchanged_ids = []

    def _normalize(value):
        # Coerce numeric values to Decimal before comparing. The Cassandra
        # driver returns float for PRICE, collect() returns Decimal, and
        # Decimal('29.99') != 29.99 in Python, so mixed types compare unequal.
        if isinstance(value, float):
            return Decimal(str(value))
        return value

    for row in src_rows:
        rid = row["ID"]
        if rid not in current_index:
            new_rows.append(row)
        else:
            current = current_index[rid]
            if any(_normalize(row[c]) != _normalize(current[c]) for c in SCD2_TRACKED_COLS):
                changed_rows.append(row)
                expired_ids.append(rid)
            else:
                unchanged_ids.append(rid)

    new_count     = len(new_rows)
    changed_count = len(changed_rows)
    unchanged_count = len(unchanged_ids)
    log.info("  New IDs:     %d", new_count)
    log.info("  Changed IDs: %d", changed_count)
    log.info("  Unchanged:   %d", unchanged_count)

    # ── Steps 1 + 2: expire old versions and insert new ones atomically ────
    # Both must land together — if the pipeline dies between them, expired
    # rows would have no IS_CURRENT successor. BEGIN/COMMIT ties them into
    # one transaction; any failure rolls back so the table is left untouched.
    rows_to_insert = new_rows + changed_rows
    if expired_ids or rows_to_insert:
        tmp = f"TMP_SCD2_EXPIRE_{run_id.replace('-', '_')}"

        # Create the temp table *before* opening the transaction — DDL issued
        # mid-transaction causes an implicit commit in Snowflake, which would
        # silently break the atomicity we're trying to establish below.
        if expired_ids:
            expire_schema = StructType([StructField("ID", StringType())])
            session.create_dataframe(
                [(eid,) for eid in expired_ids], schema=expire_schema
            ).write.mode("overwrite").save_as_table(tmp, table_type="temporary")

        try:
            session.sql("BEGIN").collect()

            if expired_ids:
                session.sql(f"""
                    MERGE INTO {fqn} AS tgt
                    USING {tmp} AS exp ON tgt.ID = exp.ID AND tgt.IS_CURRENT = TRUE
                    WHEN MATCHED THEN UPDATE SET
                        tgt.EFFECTIVE_TO = CURRENT_TIMESTAMP(),
                        tgt.IS_CURRENT   = FALSE
                """).collect()
                log.info("  Expired %d old version(s).", changed_count)

            if rows_to_insert:
                effective_now = datetime.now(timezone.utc).replace(tzinfo=None)

                scd2_records = [
                    (
                        str(uuid.uuid4()),           # SCD_KEY
                        row["ID"],
                        row["NAME"],
                        row["CATEGORY"],
                        row["PRICE"],
                        row["STOCK_QTY"],
                        row["CREATED_AT"],
                        row["PRICE_TIER"],
                        row["STOCK_VALUE"],
                        row["IS_HIGH_VALUE"],
                        row["SOURCE_SYSTEM"],
                        row["PIPELINE_RUN_ID"],
                        effective_now,               # EFFECTIVE_FROM
                        None,                        # EFFECTIVE_TO  (NULL = active)
                        True,                        # IS_CURRENT
                    )
                    for row in rows_to_insert
                ]

                scd2_schema = StructType([
                    StructField("SCD_KEY",          StringType()),
                    StructField("ID",               StringType()),
                    StructField("NAME",             StringType()),
                    StructField("CATEGORY",         StringType()),
                    StructField("PRICE",            DecimalType(10, 2)),
                    StructField("STOCK_QTY",        IntegerType()),
                    StructField("CREATED_AT",       TimestampType()),
                    StructField("PRICE_TIER",       StringType()),
                    StructField("STOCK_VALUE",      DecimalType(14, 2)),
                    StructField("IS_HIGH_VALUE",    BooleanType()),
                    StructField("SOURCE_SYSTEM",    StringType()),
                    StructField("PIPELINE_RUN_ID",  StringType()),
                    StructField("EFFECTIVE_FROM",   TimestampType()),
                    StructField("EFFECTIVE_TO",     TimestampType()),
                    StructField("IS_CURRENT",       BooleanType()),
                ])

                session.create_dataframe(scd2_records, schema=scd2_schema) \
                       .write.mode("append").save_as_table(fqn)
                log.info("  Inserted %d row(s).", len(rows_to_insert))

            session.sql("COMMIT").collect()
        except Exception:
            session.sql("ROLLBACK").collect()
            raise

    return {
        "new": new_count,
        "expired": changed_count,
        "inserted": new_count + changed_count,
        "unchanged": unchanged_count,
    }


# ── Main ──────────────────────────────────────────────────────────────────────

def main():
    records = read_from_cassandra()

    session = get_snowflake_session()
    owns_session = _INJECTED_SESSION is None
    try:
        session.use_database(SF_DATABASE)
        session.use_schema(SF_SCHEMA)
        session.use_warehouse(SF_WAREHOUSE)

        run_id = str(uuid.uuid4())

        # 1. Curated load (full snapshot, overwritten each run)
        load_curated(session, records, run_id)

        # 2. SCD Type 2 upsert
        log.info("Running SCD Type 2 ...")
        stats = apply_scd2(session, records, run_id)
        log.info("SCD2 stats: %s", stats)

        # 3. Verify: show current active rows with SCD2 columns
        log.info("Current active rows in SCD2 table:")
        session.table(f"{SF_DATABASE}.{SF_SCHEMA}.{SF_TABLE_SCD2}").filter(
            col("IS_CURRENT") == lit(True)
        ).select(
            "ID", "NAME", "PRICE", "PRICE_TIER", "EFFECTIVE_FROM", "EFFECTIVE_TO", "IS_CURRENT"
        ).show(15)

    finally:
        if owns_session:
            session.close()


if __name__ == "__main__":
    main()

"""
setup_cassandra.py
Creates a demo keyspace + products table in the local Docker Cassandra
and seeds it with sample data.

Run once before executing the pipeline:
    uv run src/setup_cassandra.py
"""

from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
import uuid
from datetime import datetime, timezone
import logging

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
log = logging.getLogger(__name__)

CASSANDRA_HOST = "127.0.0.1"
CASSANDRA_PORT = 9042
KEYSPACE = "demo_ks"


def get_cluster():
    return Cluster(
        contact_points=[CASSANDRA_HOST],
        port=CASSANDRA_PORT,
        # Uncomment if Cassandra auth is enabled:
        # auth_provider=PlainTextAuthProvider(username="cassandra", password="cassandra"),
        # No protocol_version pinned — the driver negotiates with the cluster.
        # Pin it only if you must (protocol 5 requires Cassandra 4.0+).
    )


def create_keyspace(session):
    log.info("Creating keyspace '%s' ...", KEYSPACE)
    session.execute(f"""
        CREATE KEYSPACE IF NOT EXISTS {KEYSPACE}
        WITH replication = {{'class': 'SimpleStrategy', 'replication_factor': 1}}
    """)
    session.set_keyspace(KEYSPACE)
    log.info("Keyspace ready.")


def create_table(session):
    log.info("Creating table 'products' ...")
    session.execute("""
        CREATE TABLE IF NOT EXISTS products (
            id          UUID        PRIMARY KEY,
            name        TEXT,
            category    TEXT,
            price       DECIMAL,
            stock_qty   INT,
            created_at  TIMESTAMP
        )
    """)
    log.info("Table ready.")


def seed_data(session):
    log.info("Seeding sample data ...")
    insert_stmt = session.prepare("""
        INSERT INTO products (id, name, category, price, stock_qty, created_at)
        VALUES (?, ?, ?, ?, ?, ?)
    """)
    sample_rows = [
        (uuid.uuid4(), "Laptop Pro 15",      "Electronics",  1299.99, 50,  datetime.now(timezone.utc)),
        (uuid.uuid4(), "Wireless Mouse",     "Electronics",    29.99, 200, datetime.now(timezone.utc)),
        (uuid.uuid4(), "Office Chair",       "Furniture",     349.00, 30,  datetime.now(timezone.utc)),
        (uuid.uuid4(), "Standing Desk",      "Furniture",     599.00, 15,  datetime.now(timezone.utc)),
        (uuid.uuid4(), "USB-C Hub",          "Electronics",    49.99, 120, datetime.now(timezone.utc)),
        (uuid.uuid4(), "Monitor 27inch",     "Electronics",   399.00, 40,  datetime.now(timezone.utc)),
        (uuid.uuid4(), "Mechanical Keyboard","Electronics",   129.99, 75,  datetime.now(timezone.utc)),
        (uuid.uuid4(), "Desk Lamp",          "Furniture",      39.99, 90,  datetime.now(timezone.utc)),
        (uuid.uuid4(), "Webcam 1080p",       "Electronics",    79.99, 60,  datetime.now(timezone.utc)),
        (uuid.uuid4(), "Notebook Set",       "Stationery",      9.99, 500, datetime.now(timezone.utc)),
    ]
    for row in sample_rows:
        session.execute(insert_stmt, row)
    log.info("Inserted %d rows.", len(sample_rows))


def main():
    log.info("Connecting to Cassandra at %s:%s ...", CASSANDRA_HOST, CASSANDRA_PORT)
    cluster = get_cluster()
    try:
        session = cluster.connect()
        create_keyspace(session)
        create_table(session)
        seed_data(session)
        log.info("Cassandra setup complete.")
    finally:
        cluster.shutdown()


if __name__ == "__main__":
    main()

"""
stored_proc_handler.py
Thin entry point Snowflake calls when executing the stored procedure.

Signature expected by 04_stored_proc.sql:
    run_pipeline(session, cassandra_host, cassandra_port, cassandra_keyspace)

The handler delegates all real work to pipeline.main() after patching the
pipeline module's runtime config so the core logic stays unchanged.
"""

import os


def run_pipeline(
    session,                         # Snowpark session injected by Snowflake
    cassandra_host: str,
    cassandra_port: int,
    cassandra_keyspace: str,
) -> str:
    # Override connection constants before importing pipeline so the module
    # picks up the runtime values rather than its hardcoded defaults.
    os.environ["CASSANDRA_HOST"]     = cassandra_host
    os.environ["CASSANDRA_PORT"]     = str(cassandra_port)
    os.environ["CASSANDRA_KEYSPACE"] = cassandra_keyspace

    # pipeline.main() uses the Snowpark session it creates internally, but we
    # want it to reuse the *injected* session so Snowflake manages the context.
    # Patch the module's globals directly rather than via env vars — pipeline
    # reads its config into module-level constants at import time, so if the
    # module is already cached in sys.modules (common across repeated calls
    # in the same warehouse), setting os.environ here would have no effect.
    import pipeline as p             # noqa: PLC0415  (inside zip, flat layout)
    p.CASSANDRA_HOST  = cassandra_host
    p.CASSANDRA_PORT  = cassandra_port
    p.KEYSPACE        = cassandra_keyspace
    p._INJECTED_SESSION = session    # picked up by patched get_snowflake_session

    try:
        p.main()
        return "SUCCESS"
    except Exception as exc:
        import traceback
        return f"ERROR: {exc}\n\nTraceback:\n{traceback.format_exc()}"

---Please make sure code bundle feature has been enabled in your snowflake account
CREATE OR REPLACE PROCEDURE RUN_CASSANDRA_PIPELINE(
CASSANDRA_HOST VARCHAR, CASSANDRA_PORT INT, CASSANDRA_KEYSPACE VARCHAR)
-- RUNTIME_VERSION must match the --python-version the zip was built with;
-- the cassandra-driver wheels are ABI-specific, and pipeline.py needs 3.10+.
RETURNS VARCHAR LANGUAGE PYTHON RUNTIME_VERSION = '3.11'
PACKAGES = ('snowflake-snowpark-python', 'pyarrow')
IMPORTS = ('@CASSANDRA_PIPELINE_STAGE/cassandra_pipeline.zip')
HANDLER = 'stored_proc_handler.run_pipeline'
EXTERNAL_ACCESS_INTEGRATIONS = (CASSANDRA_EAI);
CREATE OR REPLACE TASK CASSANDRA_PIPELINE_TASK
WAREHOUSE = COMPUTE_WH SCHEDULE = '60 MINUTES'
AS CALL SJAYABALDB.COCO.RUN_CASSANDRA_PIPELINE(
'cassandra.prod.internal', 9042, 'demo_ks');
 ALTER TASK CASSANDRA_PIPELINE_TASK RESUME;

Path 2 — SPCS orchestration

Snowpark Container Services runs the same Dockerfile as a job: one container, one run, then exit.You create an image repository and a compute pool (CPU_X64_XS, auto-suspend after five idleminutes), push the image with deploy/spcs/push_image.sh, and schedule EXECUTE JOB SERVICE from a Task. The container authenticates back to Snowflake with a key-pair mounted from a Snowflake secret — no password in the spec.

CREATE IMAGE REPOSITORY IF NOT EXISTS CASSANDRA_PIPELINE_REPO;
CREATE COMPUTE POOL IF NOT EXISTS CASSANDRA_PIPELINE_POOL
MIN_NODES = 1 MAX_NODES = 3
INSTANCE_FAMILY = CPU_X64_XS
AUTO_RESUME = TRUE AUTO_SUSPEND_SECS = 300;

spec:
  # Compute pools are x86_64 — build/push a linux/amd64 image to match,
  # otherwise an Apple Silicon build dies with "exec format error".

  containers:
    - name: cassandra-pipeline
      # Replace with your own registry path:
      # <org>-<account>.registry.snowflakecomputing.com/<db>/<schema>/<repo>/<image>:<tag>
      image: <account>.registry.snowflakecomputing.com/sjayabaldb/coco/cassandra_pipeline_repo/cassandra-pipeline:latest
      env:
        CASSANDRA_HOST: cassandra.prod.internal
      secrets:
        - snowflakeSecret: SJAYABALDB.COCO.CASSANDRA_PIPELINE_RSA_KEY
          directoryPath: /secrets
  restartPolicy: Never

CREATE OR REPLACE TASK CASSANDRA_SPCS_PIPELINE_TASK
WAREHOUSE = COMPUTE_WH SCHEDULE = '60 MINUTES'
AS
EXECUTE JOB SERVICE
IN COMPUTE POOL CASSANDRA_PIPELINE_POOL
EXTERNAL_ACCESS_INTEGRATIONS = (CASSANDRA_EAI)
FROM SPECIFICATION $$ ... $$;
-- ALTER TASK CASSANDRA_SPCS_PIPELINE_TASK RESUME;

# trigger_pipeline → poll_status → log_results
await ctx.query(f"""
EXECUTE JOB SERVICE
IN COMPUTE POOL CASSANDRA_PIPELINE_POOL
NAME = SJAYABALDB.COCO.{job_name}
FROM SPECIFICATION $${spec}$$
EXTERNAL_ACCESS_INTEGRATIONS = (CASSANDRA_EAI)
ASYNC
""")

Screenshots

Locally, two commands run the whole thing: the seed script once, then the pipeline as often as you like. In production the same module is either a Snowflake Task calling a stored procedure, or an SPCS job on a compute pool.

Takeaways

• Snowpark being Python is the whole advantage. The official cassandra-driver plus a typed create_dataframe() is about thirty lines of code, and it works for any source with a Python client.
• Declare your schema. An explicit StructType removes type inference, pandas, and pyarrow from the equation in one move.
• Keep pandas out of the data path. It is a transitive dependency of the connector, not a tool you need for this job.
• SCD2 is four columns, a MERGE, and a transaction. A surrogate key, two timestamps, and a boolean buy you point-in-time queries forever — but wrap the expire and the insert in BEGIN/COMMIT, or a mid-run failure leaves rows closed with no live successor.
• Productionize the same file, do not fork it. Two Snowflake-native schedules share pipeline.py: a stored-procedure Task, or an SPCS job on a compute pool.
• Snowflake cannot reach Cassandra until you say so. Network rule + External Access Integration on both the stored proc and the SPCS job. Bundle the zip for Python 3.11; the SPCS container image uses its own Python — the SCD2 logic is version-independent.The full source is still one pipeline module — plus deploy/snowflake/ for stored-proc orchestration and deploy/spcs/ for the container job. Pick the path that matches the runtime you want.

Conclusion

By unifying the agility of Apache Cassandra with the analytical power of Snowflake, this pipeline architecture enables organizations to move beyond simple data replication. By leveraging Snowpark to implement an SCD Type 2 strategy, you retain the ability to query the “current truth” of your product catalog while preserving a granular, time-stamped history of every change.

Whether you choose the lightweight efficiency of Snowflake Stored Procedures or the robust isolation of Snowpark Container Services, this approach ensures high performance and governance, transforming raw, high-velocity NoSQL data into a reliable, analytical asset.

Thanks for reading. If you have taken a different route from Cassandra into Snowflake, I would genuinely like to hear how it went.

Modernizing NoSQL Data: A Cassandra to Snowflake Integration Blueprint was originally published in Snowflake Builders Blog: Data Engineers, App Developers, AI, & Data Science on Medium, where people are continuing the conversation by highlighting and responding to this story.