从Cassandra到Snowflake：NoSQL数据集成蓝图

使用 Snowpark Python 将数据从 Apache Cassandra 迁移到 Snowflake，可以构建一条高效管道，既利用原生 Python 驱动程序，又保持 Snowflake 受治理的安全性和规模。对于需要优先维护完整变更历史（SCD Type 2）的大规模源系统，这种方法尤为有效。

Apache Cassandra（源）

Apache Cassandra 是一个分布式、宽列 NoSQL 数据库，旨在跨多台商用服务器处理大量数据。

• 数据模型： 它使用灵活的宽列模式，非常适合需要横向扩展的结构化数据。
• 使用场景： 它常用于高速应用，例如产品目录，其中价格和库存水平等属性频繁变化且需要被可靠地捕获。
• 连接性： 出于集成目的，官方的 DataStax cassandra-driver 允许 Python 应用程序直接将行读取为原生 Python 对象（元组/列表），无需中间文件格式。

Snowpark（引擎）

Snowpark 是 Snowflake 中的一组库和运行时，允许开发者使用非 SQL 语言（例如 Python）直接在 Snowflake 环境中构建数据管道。

• Python 原生： 由于 Snowpark 会话本质上是一个标准 Python 环境，整个 Python 生态系统——包括像 cassandra-driver 这样的专用数据库驱动程序——都可使用。
• DataFrame API： Snowpark 提供 DataFrame API，可将复杂的 Python 逻辑转换为单个 SQL 执行计划，然后由 Snowflake 在其高性能引擎上执行。
• 无服务器执行： 逻辑可以通过原生调度器（如 Snowflake Tasks）在 Snowflake 内部投入生产，也可以作为容器化作业通过 Snowpark Container Services（SPCS）托管运行。
• 效率： 通过声明显式模式（使用 StructType），开发者可以在数据路径中绕过对 pandas 或 pyarrow 的需求，从而减少内存开销并避免常见的类型推断错误。

高层架构

通过 Snowpark 从 Cassandra 迁移到 Snowflake，通常遵循以下结构化流程：

用例

源系统是位于 Cassandra 中的产品目录：真实环境中有大约一万个 SKU，演示键空间中有十行。价格会变化。库存水平会变化。偶尔产品会被重命名或在类别之间移动。Snowflake 中的分析师需要从中获得两样东西，而它们相互冲突：

• 当前真相 —— 每个产品一行，最新值，便于仪表板快速查询。
• 完整历史 —— 每个产品的每个版本，以及其有效的时间窗口，以便可以按截至任意日期运行利润和定价分析。简单的覆盖写入会给你第一个，却摧毁第二个。仅追加的加载会给你一堆重复数据，无法分辨哪一行是当前有效的。答案是缓慢变化维度 Type 2表：当被跟踪的属性发生变化时，通过打上 EFFECTIVE_TO 时间戳并将 IS_CURRENT 翻转为 false 来使旧行过期，然后插入新版本与之并列。

有四列被视为值得记录历史：NAME、CATEGORY、PRICE 和 STOCK_QTY。其中任何一列发生变化都会创建新版本；其他任何列的变化则不会。

技术栈

• Apache Cassandra 4.x 是源，使用 DataStax cassandra-driver 进行查询。
• Python 3.11 运行管道本身。
• Snowpark Python 提供类型化 DataFrame 和写入路径。
• Snowflake 提供存储：一张精选快照表加上一张 SCD2 历史表。
• 编排是 Snowflake 原生的：要么是由 Task 驱动的 Python 存储过程，要么是 Snowpark Container Services 上的容器作业。
• Docker 和 docker-compose 在本地为演示运行 Cassandra。
• uv 负责依赖管理。

Snowflake 设置

-- 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}")

代码片段

这是值得仔细阅读的部分。该管道是一个单一模块，包含四个可动部分：读取、类型、转换和版本化。

1 · 从 Cassandra 读取到普通元组

驱动程序返回带有真实 Python 类型的行对象——uuid.UUID、Decimal、datetime。我将 UUID 规范化为字符串，并去掉时区，以便该值干净地落入 TIMESTAMP_NTZ，然后按模式顺序返回元组列表。此时还没有 DataFrame。要完成管道的“读取”阶段，你可以使用以下函数从 Cassandra 提取数据、规范化类型（例如将 UUID 转为字符串，为 TIMESTAMP_NTZ 去掉时区），并返回一个准备好用于创建 Snowpark DataFrame 的元组列表。

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 · 声明模式，完全跳过 pandas

显式的 StructType 就是全部诀窍。Snowpark 预先获得精确类型，因此无需推断、无需 pyarrow 往返，数据路径中也没有 pandas 依赖。

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 · 使用 DataFrame API 派生业务列

五个派生列，全部以 Snowpark 表达式表示。没有任何内容在 Python 中计算——这会编译为 Snowflake 执行的单个 SQL 计划。

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 · 加载精选表

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 · 在 Python 中检测变化，而不是在 join 中

在这里，我刻意远离了 DataFrame API。DataFrame.alias() 在 Snowpark 1.54 中被标记为实验性，并且在自连接中它会破坏列名——ID 返回为 IDSRC，下一次 collect() 会因无效标识符而失败。将活动行通过一次 collect() 收集到字典中更简单，而且在这个数据量下也更快。

# 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

注意该比较双方的类型。当 collect() 读取 Snowflake 表时，PRICE 返回为 Decimal，但 Cassandra 驱动程序返回的是普通 float。在 Python 中，Decimal('29.99') != 29.99 的计算结果为 True，因此裸的 row[c] != current_index[rid][c] 检查会在每次运行时将每一行都标记为已更改，并生成幽灵 SCD2 版本。_normalize() 辅助函数在比较前将 float 强制转换为 Decimal，因此 PRICE（以及任何其他被跟踪的数值列）会以同类方式进行比较。

6 · 使用 MERGE 使旧版本过期

要退役的 ID 进入一个会话级临时表，然后由一次 MERGE 将它们标记为关闭。临时表在事务开启之前创建：在事务中途执行 DDL 会在 Snowflake 中触发隐式提交，这会悄然破坏此处建立的原子性。使用 CREATE TEMPORARY（而不是以某次运行命名的常规表）也消除了两次重叠运行互相覆盖的可能，并且它会随会话消失。

# 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 · 插入新版本

新产品和已变更产品的新版本一起插入，每个都带有新的代理键、一个开放式的有效期窗口，以及 IS_CURRENT = True。这在步骤 6 中开启的事务内运行并在结束时提交，因此过期和插入要么都成功，要么都不发生。否则，两者之间发生崩溃会留下已过期却没有任何活跃后继版本的行——某个产品的每个版本都被关闭，却没有任何版本被标记为当前。

# 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;

投入生产

配置移出源码树

主机、键空间、仓库和表名默认为演示值，并可通过环境变量覆盖。机密绝不存放在 git 中。在本地，你复制 .env.example。存储过程路径将 Cassandra 主机作为参数接收。SPCS 路径将 RSA 密钥作为 Snowflake 机密挂载到容器内。

.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

两条路径都需要到 Cassandra 的出站访问

除非你明确允许，Snowflake 无法访问 Cassandra。一个 网络规则 指定主机和端口；一个 外部访问集成（需要 ACCOUNTADMIN）附加该规则。存储过程和 SPCS 作业都声明 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;

路径 1 — Snowflake 编排

工作进程是一个 Python 存储过程。调度器是一个每小时调用它一次的 Task。代码是内部暂存区上的一个 zip 文件——pipeline.py、stored_proc_handler.py 和 cassandra-driver。Snowpark 由仓库提供；不要将其打包进去。处理程序注入 Snowflake 的会话，因此管道从不开启第二个会话。

一个会坑到你的打包细节：cassandra-driver 附带编译的 C 扩展，而仓库运行在 Linux x86_64 上。如果你在 Mac 上构建 zip，你会得到 macOS（或 arm64）的产物，存储过程会在导入时报错，而不是在部署时。应为目标平台而非本地平台解析 wheel——使用 uv 时，这意味着 uv pip install — target ./pkg — python-version 3.11 — python-platform x86_64-manylinux2014 — only-binary=:all: — no-deps cassandra-driver，而使用普通 pip 时，等价命令是 pip install — platform manylinux2014_x86_64 — only-binary=:all: — target ./pkg cassandra-driver。强制仅使用二进制解析是关键部分：它会阻止 pip 悄悄为你恰好所在的机器构建扩展。这同样适用于 SPCS 路径，其中的解决办法是用 — platform linux/amd64 构建镜像。请安装 %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;

路径 2 — SPCS 编排

Snowpark Container Services 将同一个 Dockerfile 作为作业运行：一个容器，一次运行，然后退出。你创建一个 镜像仓库 和一个 计算池（CPU_X64_XS，空闲五分钟后自动挂起），用 deploy/spcs/push_image.sh 推送镜像，并从 Task 调度 EXECUTE JOB SERVICE。容器使用从 Snowflake 机密挂载的密钥对向 Snowflake 进行身份验证——规格中没有密码。

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
""")

屏幕截图

在本地，两条命令即可运行整套流程：先运行一次种子脚本，然后按需任意次运行管道。在生产环境中，同一个模块要么是调用存储过程的 Snowflake Task，要么是计算池上的 SPCS 作业。

要点

• Snowpark 采用 Python 正是其全部优势所在。 官方的 cassandra-driver 加上类型化的 create_dataframe()，大约只需三十行代码，而且它对任何带有 Python 客户端的源都适用。
• 声明你的模式。 显式的 StructType 一举将类型推断、pandas 和 pyarrow 从等式中移除。
• 让 pandas 远离数据路径。 它是连接器的传递依赖，而不是你这项工作所需的工具。
• SCD2 就是四列、一个 MERGE 和一个事务。一个代理键、两个时间戳和一个布尔值，就能永远为你提供时间点查询——但务必把过期和插入包裹在 BEGIN/COMMIT 中，否则运行中途失败会留下已关闭却没有任何活跃后继版本的行。
• 将同一文件投入生产，不要分叉它。 两个 Snowflake 原生调度共享 pipeline.py：一个存储过程 Task，或计算池上的一个 SPCS 作业。
• 除非你明确允许，Snowflake 无法访问 Cassandra。 在存储过程和 SPCS 作业上都配置网络规则 + 外部访问集成。为 Python 3.11 打包 zip；SPCS 容器镜像使用它自己的 Python——SCD2 逻辑与版本无关。完整源码仍是一个管道模块——外加用于存储过程编排的 deploy/snowflake/ 和用于容器作业的 deploy/spcs/。选择与你想要的运行时相匹配的路径。

结论

通过将 Apache Cassandra 的敏捷性与 Snowflake 的分析能力统一起来，这一管道架构使组织能够超越简单的数据复制。通过利用 Snowpark 实现 SCD Type 2 策略，你既能保留查询产品目录“当前真相”的能力，又能保留每一次变更的细粒度、带时间戳的历史记录。

无论你选择 Snowflake 存储过程的轻量高效，还是 Snowpark Container Services 的强大隔离性，这种方法都能确保高性能和治理，将原始、高速的 NoSQL 数据转变为可靠的分析资产。

感谢阅读。如果你从 Cassandra 到 Snowflake 走了不同的路线，我非常想听听进展如何。

现代化 NoSQL 数据：Cassandra 到 Snowflake 集成蓝图最初发布于 Snowflake Builders Blog: Data Engineers, App Developers, AI, & Data Science 于 Medium，人们在那里通过高亮和回应这个故事来继续对话。