Feast 与 dbt 集成实战让 dbt 模型直接成为生产级 AI 特征【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feast在 Feast 开源特征存储The Open Source Feature Store for AI/ML中feast dbt命令组提供了一条从 dbt 模型到生产级 AI 特征的直通路径它读取dbt compile产出的manifest.json自动解析模型元数据并生成 Feast 的 Entity、DataSource 与 FeatureView 定义让数据工程师无需重写任何 SQL 变换逻辑即可把已经验证过的 dbt 模型用于实时特征服务与训练数据生成。读完本文你将掌握feast dbt import/feast dbt list的完整用法、底层实现原理、类型映射规则以及可落地的 CI/CD 实践。为什么你的 dbt 模型天生就是 AI 特征如果你已经在用 dbt那么你大概率已经完成了这些工作把原始数据转换成干净、聚合好的表为模型编写了列描述与元数据通过测试保障了数据质量把整个变换管线组织成了可维护的代码库。这些资产与 AI 特征工程的需求几乎完全重合为日报构建的聚合指标 —— 就是特征为报表 enriched 的客户属性 —— 就是特征打磨过的基于时间的计算 —— 还是特征。真正的问题从来不是模型本身而是如何把它们接入一个能提供低延迟在线服务与point-in-time 正确性防止训练阶段数据泄漏的系统中。Feast 的 dbt 集成正是为了解决这一点而设计你的 dbt 模型应该是单一事实来源single source of truth而不是在特征存储里被重新实现一遍。整个流程只需三步给想作为特征使用的 dbt 模型打上标签只需在 config 中加入tags: [feast]运行feast dbt import从 dbt 元数据自动生成 Feast 定义借助 Feast 的特征服务基础设施部署上线。你的 dbt 工作流完全不需要改变集成只是在 dbt 项目与生产 AI 服务之间架起一座桥。集成原理Feast 如何读懂你的 dbt 项目Feast 读取dbt compile生成的manifest.json从中提取来自 schema 文件的列名、类型与描述来自 dbt 模型的表位置database.schema.table你已经写好的全部元数据。随后生成 Python 代码定义 Feast 的 entities、data sources 与 feature views与你的 dbt 模型一一对应。你的文档变成特征文档数据类型变成特征类型模型变成生产就绪的特征。底层实现解析器、映射器与代码生成器从源码结构看整个集成由三个核心模块构成位于 sdk/python/feast/dbt/模块职责parser.py解析 manifest.json抽出DbtModel/DbtColumn数据类mapper.py将 dbt 元数据映射为 Feast 的 DataSource、Entity、FeatureView 对象codegen.py基于 Jinja2 模板生成可直接执行的 Python 定义文件CLI 入口在 sdk/python/feast/cli/dbt_import.py并在 sdk/python/feast/cli/cli.py 中通过cli.add_command(dbt_cmd)注册为feast dbt命令组。解析层基于dbt-artifacts-parser对 manifest 进行类型化解析支持 manifest v1-v12对应 dbt 0.19 到 1.11。值得注意的细节是dbt 可能在macros.*.supported_languages中输出javascript等值而 dbt-artifacts-parser 只接受python与sql因此 parser.py 中的_sanitize_supported_languages会先做一次清洗重试避免解析失败。端到端实战从 dbt 模型到生产特征下面以一个出行共享公司为例你已经用 dbt 构建了司机绩效指标模型现在 AI 团队要用它预测哪些司机更可能接单。全程只需六七个步骤。第 1 步安装带 dbt 支持的 Feastpip install feast[dbt]该 extra 会安装dbt-artifacts-parser。如果缺少它feast dbt相关命令会直接报错并提示Install with: pip install feast[dbt] or pip install dbt-artifacts-parser参见 parser.py 与集成测试 tests/integration/dbt/test_dbt_integration.py。第 2 步给现有 dbt 模型打一个标签你已有的计算司机指标的模型只需加一个标签即可标记给 Feast-- models/features/driver_features.sql {{ config( materializedtable, tags[feast] -- ← 只需加这个标签 ) }} WITH driver_stats AS ( SELECT driver_id, DATE(completed_at) as date, AVG(rating) as avg_rating, COUNT(*) as total_trips, SUM(fare_amount) as total_earnings, AVG(trip_duration_minutes) as avg_trip_duration FROM {{ ref(trips) }} WHERE status completed GROUP BY driver_id, DATE(completed_at) ) SELECT driver_id, TIMESTAMP(date) as event_timestamp, avg_rating, total_trips, total_earnings, avg_trip_duration, CASE WHEN total_trips 5 THEN true ELSE false END as is_active FROM driver_stats一个标签而已。你的模型继续照常为分析负载服务行为完全不变。第 3 步复用你已有的文档Feast 直接消费你为 dbt 模型编写的 schema 文档零重复# models/features/schema.yml version: 2 models: - name: driver_features description: Daily aggregated features for drivers including ratings and activity metrics columns: - name: driver_id description: Unique identifier for the driver data_type: STRING - name: event_timestamp description: Date of the feature computation data_type: TIMESTAMP - name: avg_rating description: Average rating received from riders data_type: FLOAT64 - name: total_trips description: Total number of completed trips data_type: INT64 - name: total_earnings description: Total earnings in dollars data_type: FLOAT64 - name: avg_trip_duration description: Average trip duration in minutes data_type: FLOAT64 - name: is_active description: Whether driver completed 5 trips (active status) data_type: BOOLEAN列描述和数据类型会自动成为 Feast 中的特征文档——写一次处处复用。第 4 步正常编译 dbt 项目cd your_dbt_project dbt compile这会生成target/manifest.json包含你全部模型的元数据——这正是你已经产出的那份产物。第 5 步查看 Feast 发现了什么用 Feast CLI 发现打了标签的模型feast dbt list -m target/manifest.json --tag feast输出形如Found 1 model(s): driver_features [tags: feast] Table: my_project.my_dataset.driver_features Description: Daily aggregated features for drivers including ratings and activity metricsfeast dbt list还支持--show-columns参数展示每个模型的列级类型细节。第 6 步把 dbt 模型导入 Feast现在就是核心的“魔法”时刻——从 dbt 模型自动生成生产级特征定义feast dbt import -m target/manifest.json \ --entity-column driver_id \ --data-source-type bigquery \ --tag feast \ --output feature_repo/driver_features.py数秒之内Feast 就会基于你现有的 dbt 模型生成一个完整的 Python 文件包含生产 AI 服务所需的一切 Feast feature definitions generated from dbt models. Source: target/manifest.json Generated by: feast dbt import from datetime import timedelta from feast import Entity, FeatureView, Field from feast.types import Bool, Float64, Int64, String from feast.infra.offline_stores.bigquery_source import BigQuerySource # Entities driver_id Entity( namedriver_id, join_keys[driver_id], descriptionEntity key for dbt models, tags{source: dbt}, ) # Data Sources driver_features_source BigQuerySource( namedriver_features_source, tablemy_project.my_dataset.driver_features, timestamp_fieldevent_timestamp, descriptionDaily aggregated features for drivers including ratings and activity metrics, tags{dbt.model: driver_features, dbt.tag.feast: true}, ) # Feature Views driver_features_fv FeatureView( namedriver_features, entities[driver_id], ttltimedelta(days1), schema[ Field(nameavg_rating, dtypeFloat64, descriptionAverage rating received from riders), Field(nametotal_trips, dtypeInt64, descriptionTotal number of completed trips), Field(nametotal_earnings, dtypeFloat64, descriptionTotal earnings in dollars), Field(nameavg_trip_duration, dtypeFloat64, descriptionAverage trip duration in minutes), Field(nameis_active, dtypeBool, descriptionWhether driver completed 5 trips (active status)), ], onlineTrue, sourcedriver_features_source, descriptionDaily aggregated features for drivers including ratings and activity metrics, tags{dbt.model: driver_features, dbt.tag.feast: true}, )从生成的代码可以看到Feast 为 DataSource 和 FeatureView 都自动打上了dbt.model与dbt.tag.feast溯源标签方便后续追踪每个特征的来源。代码生成模板位于 sdk/python/feast/dbt/codegen.pyFEAST_FILE_TEMPLATE支持 BigQuery、Snowflake、File 三种数据源并会自动按需收集类型 import包括Array(Int64)这类嵌套类型的 import 组合。第 7 步应用到你的特征存储之后就可以使用标准的 Feast 命令物化这些特征cd feature_repo feast apply feast materialize-incremental $(date -u %Y-%m-%dT%H:%M:%S)发生了什么你刚刚在没有重写一行变换逻辑的情况下把 dbt 模型带到了生产级 AI 特征的位置。你的 dbt 模型——包括精心编写的 SQL、文档和测试——现在能以毫秒级延迟服务实时预测通过 point-in-time 正确性防止训练阶段的数据泄漏随着 dbt 模型更新自动与数仓同步使用你已经写好的描述实现自文档化。更重要的是当你更新 dbt 模型新增列或调整逻辑后只需重新运行feast dbt import和feast apply生产特征就能与 dbt 这个事实来源保持同步。CLI 参数详解与底层校验逻辑feast dbt import的完整参数由 sdk/python/feast/cli/dbt_import.py 中的 Click 选项定义参数简写默认值说明--manifest-path-m必填dbtmanifest.json路径通常为target/manifest.json--entity-column-e必填实体列名可多次指定如-e user_id -e merchant_id--data-source-type-dbigquery数据源类型可选bigquery、snowflake、file--timestamp-field-tevent_timestamp用于 point-in-time join 的时间戳列--tag—无只导入带指定 dbt 标签的模型如--tag feast--model—无指定要导入的具体模型名可多次指定--ttl-days—1FeatureView 的 TTL天--dry-run—False预览将创建的内容而不应用变更--exclude-columns—无从特征中排除的列逗号分隔--output-o无输出 Python 文件路径指定后生成代码而非写入 registry从源码可以看到命令执行时的多层校验与容错逻辑这些在集成测试 tests/integration/dbt/test_dbt_integration.py 与单元测试 tests/unit/dbt/test_mapper.py、tests/unit/dbt/test_parser.py 中均有覆盖manifest 缺失抛出FileNotFoundError提示先运行dbt compile或dbt runJSON 无效抛出ValueError并建议dbt clean dbt compile实体列为空或重复分别报错并退出时间戳列缺失该模型被警告并跳过实体列缺失该模型被警告并跳过没有匹配模型给出标签/模型名过滤条件并正常退出未安装 dbt-artifacts-parser抛出ImportError并提示安装方式。一个关键行为是--output与直接 apply 是两种互斥的落地方式。指定--output时只生成 Python 文件并提示“You can now import this file in your feature_store.yaml repo”不指定时则走store.apply(all_objects)直接写入 registry此时需要在有效的 feature repo 目录下运行命令内部会调用cli_check_repo与create_feature_store。类型映射从数据库类型到 Feast 类型dbt 模型列的数据类型会被自动映射为 Feast 类型映射表位于 sdk/python/feast/dbt/mapper.py 的DBT_TO_FEAST_TYPE_MAP覆盖 BigQuery、Snowflake、Redshift、PostgreSQL 及常见 SQL 类型字符串STRING、TEXT、VARCHAR、CHAR、NVARCHAR等 →String整数INT/INTEGER/BIGINT/INT64→Int64SMALLINT/TINYINT/INT32→Int32浮点FLOAT/REAL/FLOAT32→Float32DOUBLE/FLOAT64/DOUBLE PRECISION→Float64布尔BOOL/BOOLEAN→Bool时间TIMESTAMP含_NTZ/_LTZ/_TZ、DATETIME、DATE、TIME→UnixTimestamp二进制BYTES/BINARY/VARBINARY/BLOB→Bytes。映射还处理了几类复杂情况均有单元测试覆盖参数化类型VARCHAR(255)、CHAR(10)会先剥离括号参数再匹配基础类型SnowflakeNUMBER(precision, scale)scale 0→Float64precision ≤ 9→Int32precision ≤ 18→Int64precision 18→Float64防止超出 Int64 范围数组类型ARRAYSTRING、ARRAYINT64映射为Array(元素类型)只支持原始类型复杂嵌套结构回退为Array(String)未知类型与空类型安全回退为String。映射结果同时决定了 Entity 的ValueType通过FEAST_TYPE_TO_VALUE_TYPE推断见 mapper.py例如driver_id INT64会被推断为ValueType.INT64。高级用法多实体支持对于涉及多个实体的特征如用户-商户交易可多次指定实体列feast dbt import -m target/manifest.json \ -e user_id \ -e merchant_id \ --tag feast \ -o feature_repo/transaction_features.py这会生成带复合键的 FeatureView适用于以用户和商户为键的交易特征推荐系统中的交互特征多对多关系特征。集成测试 tests/integration/dbt/test_dbt_integration.py 中的TestMultiEntityColumns验证了多实体 FeatureView 的生成与可执行性并校验了“实体列数量与实体对象数量不匹配”时会抛出ValueError。Snowflake 与其他数据源集成当前支持 BigQuery、Snowflake 与文件类Parquet 等三种后端Snowflakefeast dbt import -m manifest.json \ -e user_id \ -d snowflake \ -o features.py生成的代码会使用SnowflakeSource并分别用模型的database、schema、alias填充database/schema/table参数。文件类数据源Parquet 等feast dbt import -m manifest.json \ -e user_id \ -d file \ -o features.py注意文件数据源当前以模型名生成占位路径如/data/product_features.parquet集成测试中对应断言data_source.path /data/product_features.parquet实际使用时需要按你的文件布局调整。自定义生成的代码你可以用更多选项微调导入结果feast dbt import -m target/manifest.json \ -e driver_id \ -d bigquery \ --timestamp-field created_at \ --ttl-days 7 \ --exclude-columns internal_id,temp_field \ -o features.py--exclude-columns会把指定列从 FeatureView schema 中剔除时间戳列总是被排除--ttl-days控制特征在在线存储中的存活时间。最佳实践1. 建立标签约定利用 dbt 的配置层级自动给整个目录打标签# dbt_project.yml models: my_project: features: tags: [feast] # features/ 下所有模型自动打标2. 维护丰富文档dbt schema 文件中的列描述会成为 Feast 中的特征描述形成一个自文档化的特征目录。投入时间维护 dbt 模型文档会在特征可发现性上获得回报。3. 集成 CI/CD在部署流水线中自动化特征定义的更新# .github/workflows/features.yml name: Update Features on: push: paths: - dbt_project/** jobs: update-features: runs-on: ubuntu-latest steps: - uses: actions/checkoutv4 - name: Setup Python uses: actions/setup-pythonv5 with: python-version: 3.11 - name: Install dependencies run: | pip install feast[dbt] pip install dbt-bigquery - name: Compile dbt run: | cd dbt_project dbt compile - name: Generate Feast definitions run: | feast dbt import -m dbt_project/target/manifest.json \ -e user_id \ -d bigquery \ --tag feast \ -o feature_repo/features.py - name: Apply to feature store run: | cd feature_repo feast apply4. 用 Dry Run 做验证生成代码前先预览将要创建的内容feast dbt import -m manifest.json -e driver_id --dry-run这有助于在提交前发现缺列、类型不正确等问题——命令只会打印摘要与校验警告不会产生任何实际变更。5. 对生成的代码做版本控制把生成的 Python 文件提交到仓库可以带来特征定义变更的追踪dbt 到 Feast 映射的代码评审可见性需要时的回滚能力。当前限制与未来路线图dbt 集成当前处于 alpha 阶段存在一些已知限制源码与文档均确认数据源支持当前仅支持 BigQuery、Snowflake 与文件类数据源CLI 的--data-source-type只接受这三种取值传入其他值会抛出ValueError手动指定实体必须显式指定实体列尚不能自动推断无增量更新每次导入都会生成完整文件。从代码与文档可以推断社区正在推进的增强方向包括从外键关系自动推断实体、支持更多数据源如 Redshift、Postgres、增量更新以保留自定义修改以及针对复杂嵌套结构的增强类型映射。结语你的 dbt 模型值得拥有生产级 AI你已经为 dbt 模型投入了大量时间——它们干净、有文档、经过测试、被组织信任。它们不应该为了支撑 AI 而被重写而应该原样工作。Feast 的 dbt 集成让这成为可能。你的 dbt 模型变成生产级 AI 特征时✅ 无需重写或重复建设✅ dbt 工作流零改动✅ 全部文档得到保留✅ 支持实时预测服务✅ 训练阶段保持 point-in-time 正确性。如果你是一个 dbt 用户并且刚被要求“让这些模型为 AI 服务”这就是你的答案。快速上手pip install feast[dbt] cd your_dbt_project dbt compile feast dbt import -m target/manifest.json -e your_entity_column -d bigquery想深入了解实现可以继续阅读本仓库中的相关源码CLI 入口 sdk/python/feast/cli/dbt_import.py、manifest 解析 sdk/python/feast/dbt/parser.py、对象映射 sdk/python/feast/dbt/mapper.py、代码生成 sdk/python/feast/dbt/codegen.py以及覆盖完整工作流的集成测试 sdk/python/tests/integration/dbt/test_dbt_integration.py。【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feast创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考