构建MLOps批处理管道：Feast+Cassandra存特征需中间持久化吗？

Feast原生支持Spark DataFrame作为输入数据源，且Apache Cassandra是Feast官方兼容的离线存储后端，因此可以直接将Spark处理完成的DataFrame通过Feast的Spark SDK写入Cassandra，完全跳过中间持久化环节，减少IO开销、数据冗余和流程复杂度。

直接对接的实现示例 #

from feast import FeatureStore
from feast.infra.offline_stores.cassandra import CassandraOfflineStoreConfig
from pyspark.sql import SparkSession

# 初始化Spark会话
spark = SparkSession.builder.appName("FeastCassandraWriter").getOrCreate()

# 从Hive读取并处理数据，得到规整后的特征DataFrame
processed_features_df = spark.sql("""
    SELECT user_id, transaction_amount, avg_spend_30d 
    FROM hive.raw_transactions 
    WHERE event_timestamp >= '2024-01-01'
""")

# 初始化Feast Store，配置Cassandra离线存储
store = FeatureStore(
    config={
        "project": "mlops_pipeline",
        "offline_store": CassandraOfflineStoreConfig(
            hosts=["cassandra-node-1", "cassandra-node-2"],
            keyspace="feast_features",
            username="cass_user",
            password="cass_pass"
        )
    }
)

# 定义FeatureView（也可提前在Feast repo中预定义）
from feast import FeatureView, Field
from feast.types import Float32, Int64
from feast.entity import Entity

user_entity = Entity(name="user_id", value_type=Int64)

transaction_features_view = FeatureView(
    name="transaction_features",
    entities=[user_entity],
    ttl=86400*365,
    schema=[
        Field(name="transaction_amount", dtype=Float32),
        Field(name="avg_spend_30d", dtype=Float32)
    ],
    online=False
)

# 直接将Spark DataFrame写入Feast的Cassandra离线存储
store.materialize_to_offline_store(
    feature_views=[transaction_features_view],
    source=processed_features_df,
    start_time="2024-01-01",
    end_time="2024-06-01"
)

仅在以下场景考虑中间持久化 #

• 合规/审计需求：如果行业监管要求必须留存处理后特征的原始快照，或需要支持数据回滚重跑，可以将处理后数据暂存到Hive/S3等低成本存储
• 作业解耦：若Spark数据处理作业的执行频率与Feast特征写入频率不一致（比如Spark每天跑但Feast每周批量写入），或需要多下游系统（如BI、其他ML管道）复用处理后数据
• 极端数据规模：当处理PB级数据且Cassandra写入资源受限，可先将数据落地到分布式存储，再通过Feast分批次读取写入Cassandra，避免长时间占用Cassandra集群资源

性能优化建议 #

• 按Cassandra的分区键设置Spark DataFrame的分区，避免写入热点
• 调整Spark executor资源（spark.executor.instances/spark.executor.cores）和Feast的批量写入参数（batch_size），匹配Cassandra的写入能力
• 若需临时落地数据，启用Spark的列式存储优化（如Parquet），减少存储成本和读取耗时

内容的提问来源于stack exchange，提问作者Eya