Feast Trino Offline Store 完整指南配置、点对点时间连接与本地测试【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feast本文以 Feast 仓库中 sdk/python/feast/infra/offline_stores/contrib/trino_offline_store 包对应 Sphinx API 文档 feast.infra.offline_stores.contrib.trino_offline_store为核心系统讲解如何把 Trino 作为 Feast 的离线存储Offline Store使用包括离线存储配置项与五种认证方式、TrinoSource 数据源声明、历史特征检索点对点时间连接的底层 SQL 模板与执行链路、类型映射规则以及基于 Docker 的本地端到端测试方法。读完本文你将掌握在feature_store.yaml中完整配置 Trino 离线存储、编写 Trino 数据源定义并复现官方集成测试的完整能力。模块总览Trino 离线存储的代码结构Trino 离线存储是 Feast 众多社区贡献contrib离线存储实现之一位于sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/目录下由以下核心模块组成模块职责trino.pyTrinoOfflineStoreConfig配置模型、TrinoOfflineStore离线存储实现、TrinoRetrievalJob检索任务trino_queries.py对 Trino Python 客户端trino.dbapi的轻量封装Trino、Query、Results、QueryStatustrino_source.pyTrinoSource数据源定义、TrinoOptions、SavedDatasetTrinoStorage数据集持久化trino_type_map.pyTrino 类型 ↔ FeastValueType↔ PyArrow 类型的双向映射connectors/upload.py将 Pandas DataFrame 批量上传为 Trino 临时表test_config/manual_tests.py本地手工测试的 repo 配置tests/集成测试data_source.py、test_trino_queries.pySphinx 文档 feast.infra.offline_stores.contrib.trino_offline_store.rst 是这些模块的 API 自动文档入口其中automodule指令直接引用上述trino、trino_queries、trino_source、trino_type_map四个模块并包含connectors、test_config、tests三个子包。本文即围绕这套模块体系展开。Trino 离线存储配置feature_store.yaml 完整参数在feature_store.yaml中通过offline_store一节启用 Trino 离线存储type固定为trino。核心配置模型是 trino.py 中的TrinoOfflineStoreConfig参数如下project: my_project registry: data/registry.db provider: local offline_store: type: trino # 离线存储类型选择器固定为 trino host: localhost # Trino 集群主机 port: 8080 # Trino 集群端口 catalog: hive # Trino catalog dataset: ci # 临时表所在 Trino datasetschema user: user # 连接 Trino 的用户名 source: trino-python-client # 可选的 Trino 客户端 source 标识便于调试 http-scheme: http # 可选http 或 https默认 http ssl-verify: true # 可选是否校验 SSL 证书默认 true connector: # 必填Trino connector 类型及附加参数 type: hive # 至少需要 type 字段 file_format: parquet # 例如 hive connector 需要文件格式各字段说明默认值与别名均来自TrinoOfflineStoreConfig源码type默认trino离线存储类型选择器host/portTrino 集群的连接地址必填catalogTrino catalog必填user连接用户必填source默认trino-python-client作为请求中携带的客户端来源标识方便在 Trino 侧调试http-scheme默认http仅支持http/https二选一ssl-verify默认true对应参数别名ssl-verify控制是否校验 Trino 集群下发的 SSL 证书x-trino-extra-credential-header可选指定X-Trino-Extra-CredentialHTTP 头例如user1pwd1, user2pwd2connector必填字典决定临时表通过哪个 Trino connector 创建type至少给出例如{type: bigquery}或{type: hive, file_format: parquet}dataset默认feast用于存放临时实体表entity table的 Trino datasetauth可选认证配置支持kerberos、basic、jwt、oauth2、certificate五种方式。五种认证方式详解认证由 trino.py 中的AuthConfig及各认证模型实现最终通过to_trino_auth()映射到trino.auth的认证类BasicAuthentication、KerberosAuthentication、JWTAuthentication、OAuth2Authentication、CertificateAuthentication。basic用户名密码auth: type: basic config: username: my_user password: my_password对应BasicAuthModelusername、password均必填。kerberos对应KerberosAuthModel支持以下参数括号内为 YAML 别名configconfig-filekrb5 配置文件路径可选service_nameservice-name服务名可选mutual-authentication是否启用双向认证默认falseforce-preemptive是否强制预发认证默认falsehostname-override主机名覆盖可选sanitize-mutual-error-response默认trueprincipalKerberos principal可选delegate默认falseca-bundle-fileCA 证书包路径可选。auth: type: kerberos config: principal: feastEXAMPLE.COM service-name: trinojwt对应JWTAuthModeltoken字段为必填以SecretStr保存避免明文泄露。auth: type: jwt config: token: your-jwt-tokenoauth2不需要config。AuthConfig的校验逻辑config_only_nullable_for_oauth2规定除oauth2外其余认证类型必须提供config否则抛出ValueError。auth: type: oauth2certificate对应CertificateAuthModelcert-file与key-file均为可选路径。auth: type: certificate config: cert-file: /path/to/client.crt key-file: /path/to/client.keyto_trino_auth()的实现要点根据type从CLASSES_BY_AUTH_TYPE取出认证类与配置模型oauth2直接实例化其余类型将config传入对应模型构造并把SecretStr字段还原为明文后再传给 Trino 认证类。连接客户端的建立无论配置还是检索最终都会调用 trino.py 的_get_trino_client(config)构造统一的Trino客户端若配置了auth先通过to_trino_auth()得到认证对象再连同host、port、user、catalog、source、http_scheme、verify、extra_credential一并传入。数据源声明TrinoSource 与 TrinoOptions要基于 Trino 表或 SQL 定义特征数据使用 trino_source.py 中的TrinoSource。它与DataSource基类一样支持以下参数name、timestamp_field、table、created_timestamp_column、field_mapping、query、description、tags、owner其中table与query必须且只能指定一个两者都为空时抛出DataSourceNoNameException。典型用法特征仓库定义文件from feast import FeatureView, Field from feast.infra.offline_stores.contrib.trino_offline_store.trino_source import ( TrinoSource, ) from feast.types import Float32, Int64 driver_hourly_stats TrinoSource( namedriver_hourly_stats, tablehive.default.driver_hourly_stats, # 方式一直接引用表 timestamp_fieldevent_timestamp, created_timestamp_columncreated, ) # 也可以使用 SQL 查询作为数据源 driver_stats_query TrinoSource( namedriver_stats_query, querySELECT * FROM hive.default.driver_hourly_stats WHERE region us, timestamp_fieldevent_timestamp, ) driver_stats_fv FeatureView( namedriver_hourly_stats, entities[driver_id], schema[ Field(nameconv_rate, dtypeFloat32), Field(nameacc_rate, dtypeFloat32), Field(namedriver_id, dtypeInt64), ], sourcedriver_hourly_stats, ttltimedelta(hours6), )TrinoSource的关键机制source_type()返回DataSourceProto.BATCH_TRINO_to_proto_impl()将table/query序列化进trino_options字段from_proto()负责反向还原从而支持注册表registry的持久化get_table_query_string()返回table或query本身作为后续所有 SQL 中引用该数据源的FROM表达式get_table_column_names_and_types()通过SELECT * FROM {table} LIMIT 1或SELECT * FROM ({query}) LIMIT 1探测列名与类型validate()直接调用上述方法在feast apply阶段即校验表/查询是否可访问source_datatype_to_feast_value_type()将列类型映射为 FeastValueType见后文类型映射。特征检索执行链路Trino 客户端封装客户端、查询与结果对象trino_queries.py 对trino.dbapi做了三层封装Trino懒加载游标_get_cursor()首次调用时通过trino.dbapi.connect(...)建立连接并携带X-Trino-Extra-Credential头提供execute_query(query_text)同步执行并返回Results以及create_query(query_text)返回尚未执行的QueryQuery管理查询生命周期状态机为QueryStatusPENDING → RUNNING → COMPLETED/ERROR/CANCELLED。在主线程中执行时会注册SIGINT/SIGTERM信号处理以支持cancel()execute()记录执行耗时、fetchall()取回数据出错时抛出TrinoQueryError并将状态置为ERROR无论成败最终都会关闭游标Results持有原始行数据与列元信息提供columns_names、schema列名 → Trino 类型、pyarrow_schema经类型映射转换为 PyArrow Schema与to_dataframe()timestamp*列自动pd.to_datetime空值填充np.nan。检索任务 TrinoRetrievalJobTrinoRetrievalJobtrino.py是所有离线检索的统一返回类型提供to_df()/to_arrow()内部调用_to_df_internal()/_to_arrow_internal()同步执行查询并把结果转为 Pandas DataFrame / PyArrow Table含 on-demand 转换to_sql()直接返回将要执行的 SQL 文本便于调试to_trino(destination_table, timeout1800, retry_cadence10)把历史特征查询结果落成 Trino 表表名默认形如{catalog}.{dataset}.historical_{YYYYMMDD}_{7位随机id}实际执行CREATE TABLE {dest} AS ({query})persist(storage)配合SavedDatasetTrinoStorage将结果持久化为已保存数据集dataset否则抛出ValueError临时表管理构造时传入temp_table后_drop_temp_table()负责在查询结束、异常或对象销毁__del__时执行DROP TABLE IF EXISTS防止脏数据残留。点对点时间连接离线取数的核心 SQLpull_latest_from_table_or_query物化最新特征物化materialization场景调用pull_latest_from_table_or_querytrino.py对每个 join key 分区按时间戳及可选的created_timestamp_column降序取第一行即每个实体最新的一条特征记录SELECT field_string FROM ( SELECT field_string, ROW_NUMBER() OVER(PARTITION BY join_key ORDER BY timestamp_field DESC, created_timestamp_column DESC) AS _feast_row FROM from_expression WHERE timestamp_field BETWEEN TIMESTAMP start_date AND TIMESTAMP end_date ) WHERE _feast_row 1注意当特征视图没有实体时会自动附加DUMMY_ENTITY_VAL AS DUMMY_ENTITY_IDDUMMY_ENTITY_ID/DUMMY_ENTITY_VAL来自 feature_view.py使无实体特征视图也能物化。get_historical_features训练数据集生成训练数据获取走get_historical_featurestrino.py完整流程为构造临时表引用{catalog}.{dataset}.{临时表名}_get_table_reference_for_new_entity临时表名来自offline_utils.get_temp_entity_table_name()若entity_df是 Pandas DataFrame通过upload_pandas_dataframe_to_trino上传为 Trino 临时表并推断实体表 schema若entity_df是 SQL 字符串则以子查询({entity_df})形式直接引用并用SELECT * FROM ({entity_df}) LIMIT 1探测 schema_upload_entity_df_and_get_entity_schema推断事件时间戳列infer_event_timestamp_from_entity_df计算实体时间戳范围[min, max]_get_entity_df_event_timestamp_range其中字符串实体 DataFrame 的时间戳列会被pd.to_datetime(..., utcTrue)规范化校验实体 DataFrame 包含所有预期 join keyassert_expected_columns_in_entity_df调用offline_utils.get_feature_view_query_context构建查询上下文再以MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN为模板经offline_utils.build_point_in_time_query渲染出最终点对点时间连接 SQL返回携带temp_table、metadata特征列表、key 列表、事件时间戳范围的TrinoRetrievalJob。TrinoOfflineStore.supports_filter_by_created_timestamp True声明该实现支持按created_timestamp过滤配合filter_by_created_timestamp参数生效。点对点时间连接 SQL 模板解析MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOINtrino.py是核心 SQL 模板采用 Jinja 语法渲染通过多个 CTE 分步实现时序正确性entity_dataframe为实体表追加entity_timestamp别名并为每个特征视图计算{fv}__entity_row_unique_id由实体列与实体事件时间戳拼接的字符串作为跨 CTE 关联的确定性唯一键{fv}__entity_dataframe按实体列 事件时间戳 唯一 id 去重GROUP BY{fv}__subquery从特征表/查询中选出事件时间戳、created_timestamp若配置、实体列与特征列并限制timestamp_field from_iso8601_timestamp(max_event_timestamp)且当 TTL 非 0 时限制 min_event_timestamp特征列名依据full_feature_names与field_mapping决定是否加特征视图名前缀{fv}__base与实体表INNER JOIN条件为event_timestamp entity_timestampTTL 非 0 时追加event_timestamp entity_timestamp - interval {ttl} secondTTL 以秒为单位启用filter_by_created_timestamp时再约束created_timestamp entity_timestamp最后按实体列等值匹配{fv}__dedup / {fv}__latest当配置了created_timestamp_column先按唯一 id 事件时间戳取MAX(created_timestamp)去重再用ROW_NUMBER() OVER(PARTITION BY 唯一id ORDER BY event_timestamp DESC, created_timestamp DESC)取每个实体的最新特征{fv}__cleaned把最新行 join 回 base得到每个实体-时间点的最终特征行最后以entity_dataframe为基准对各特征视图做LEFT JOIN ... USING ({fv}__entity_row_unique_id)拼出完整训练表。这套模板与 Feast 其他离线存储共享同一套offline_utils.build_point_in_time_query语义但针对 Trino 方言做了适配如from_iso8601_timestamp、interval {ttl} second、ARRAY字面量等。pull_all_from_table_or_query全量拉取用于feature_store.materialize之外的全量取数如数据集校验按时间戳范围过滤后返回全部行trino.py时间戳过滤 SQL 由offline_utils.get_timestamp_filter_sql生成使用cast_styletimestamp的 Trino 风格转换。类型映射Trino、Feast 与 PyArrow 三向转换trino_type_map.py 提供三类转换Trino → FeastValueTypetrino_to_feast_value_type基本映射为tinyint/smallint/int/integer → INT32、bigint → INT64、double/real → DOUBLE/FLOAT、timestamp → UNIX_TIMESTAMP、char/varchar/date/binary/varbinary/json → STRING、boolean → BOOL带参类型做归一化decimal(p,s)按精度p是否大于 32 分别映射为FLOAT/DOUBLEtimestamp(...)、varchar(...)、char(...)前缀归一化未支持类型直接抛ValueError。PyArrow → Trinopa_to_trino_value_typelist...映射为array...date→datetimestamp/timestamp with time zone区分有无tzdecimal128(p,s)/decimal256(p,s)归一化为decimal(p,s)map、struct、large_string降级为varchar数值与字符串映射见type_map如int32 → int、float → double、string → varchar。Trino → PyArrowtrino_to_pa_value_typearray(...)递归映射为pa.list_decimal(p,s)按精度映射pa.float32/pa.float64timestamp*→pa.timestamp(us)varchar/char/row(...)/map(...)→pa.string()其余查_TRINO_TO_PA_TYPE_MAP。这一转换用于Results.pyarrow_schema是检索结果元数据与 Arrow 导出的基础。实体表上传Pandas DataFrame → Trino 临时表connectors/upload.py 实现upload_pandas_dataframe_to_trino用于把实体 DataFrame 上传为 Trino 临时表。其 connector 兼容性逻辑源码注释给出了完整 YAML 示例offline_store: type: trino host: localhost port: 8080 catalog: hive dataset: ci connector: type: hive file_format: parquet处理规则按 connectortype分支不支持 CREATE TABLE 的 connectorCONNECTORS_DONT_SUPPORT_CREATE_TABLEdruid、elasticsearch、googlesheets、jmx、kafka、kinesis、localfile、pinot、postgresql、prometheus、redis、thrift、tpcds、tpch、qdrant直接抛ValueError无需 WITH 语句的 connectorCONNECTORS_WITHOUT_WITH_STATEMENTSbigquery、cassandra、memory、mongodb、mysql、oracle、redshift、memsql、lakehouseWITH 子句为空hive / iceberg必须提供file_format参数生成WITH (format {file_format})否则抛ValueErrorkudu / phoenix / sqlserver明确标注“尚未支持欢迎 PR”其他未列出的 connector 抛ValueError。上传流程分两步先用CREATE TABLE IF NOT EXISTS {table} ({schema}) {with_statement}建表schema 由pyarrow_schema_from_dataframe将 DataFrame 的 PyArrow schema 逐列转为 Trino 类型再按默认batch_size1000000100 万行分批执行INSERT INTO {table} ({columns}) VALUES {values}。值格式化format_pandas_row细节时间戳列包装为TIMESTAMP ...带时区时先转 UTC格式%Y-%m-%d %H:%M:%S.%flist/ndarray/tuple转ARRAY[...]字符串加单引号NaN/None转NULL其余原样输出。本地开发与测试Docker 起 Trino 跑集成测试仓库在根目录 Makefile 中提供了完整的本地测试命令READMEsdk/python/feast/infra/offline_stores/contrib/trino_offline_store/README.md对其做了说明。1. 启动本地 Trino 容器默认版本TRINO_VERSION ? 376make start-trino-locally等价于docker run --detach --rm -p 8080:8080 --name trino \ -v sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/test_config/properties/:/etc/catalog/:ro \ trinodb/trino:376启动后 Trino 监听 8080 端口catalog 配置目录来自test_config/properties/内含memory.properties即测试用的 memory catalog。测试执行期间的查询可通过 http://0.0.0.0:8080/ui 的本地集群 UI 观察便于调试插件生成的 SQL。2. 运行 universal 测试套件make test-trino-plugin-locally实际执行cd sdk/python \ FULL_REPO_CONFIGS_MODULEfeast.infra.offline_stores.contrib.trino_offline_store.test_config.manual_tests \ IS_TESTTrue \ python -m pytest --integration tests/manual_tests.py注册了IntegrationTestRepoConfig(providerlocal, offline_store_creatorTrinoSourceCreator)即以本地 provider Trino 离线存储跑通用测试套件。3. 停止并清理容器make kill-trino-locally等价于docker stop trino。也可用docker ps查看容器后用docker stop {NAME/SHA}手动停止。测试基建细节tests/data_source.py 展示了官方测试的组织方式可作为自建测试的参照trino_containersession 级 fixture基于testcontainers启动trinodb/trino:376挂载catalog目录到/etc/catalog/暴露 8080 端口等待日志中出现SERVER STARTED30 秒超时TrinoSourceCreator创建Trino客户端useruser、catalogmemory、http_schemehttp测试数据源时执行CREATE SCHEMA IF NOT EXISTS memory.{project}、DROP TABLE IF EXISTS后经upload_pandas_dataframe_to_trino灌入数据并用create_offline_store_config()返回TrinoOfflineStoreConfig(host..., port..., catalogmemory, connector{type: memory}, ...)供测试框架使用。此外tests/test_trino_queries.py 针对Trino客户端封装编写单元测试可与集成测试配合验证查询执行与结果解析逻辑。通用 Trino 集成测试的入口则位于 Makefile 的test-python-universal-trino通过FULL_REPO_CONFIGS_MODULEsdk.python.feast.infra.offline_stores.contrib.trino_repo_configuration加载配置其中默认组合为 local provider Trino 离线存储 Redis 在线存储见 trino_repo_configuration.py。在仓库其他位置的集成点Trino 离线存储在 Feast 整体架构中并非孤立模块以下位置与之直接相关repo_config.pyOfflineStoreConfig的解析入口type: trino即据此实例化TrinoOfflineStoreConfigdata_source.pyDataSource基类与 protobuf 序列化TrinoSource继承其协议dataset_utils.py数据集SavedDataset相关工具与SavedDatasetTrinoStorage.persist配合Sphinx API 文档 feast.infra.offline_stores.contrib.trino_offline_store.rst 及对应的 connectors、test_config、tests 子包文档页是查看全部公开 APIautomodule展开的入口。整体调用关系可以概括为feature_store.yaml配置TrinoOfflineStoreConfig→TrinoOfflineStore各静态方法pull_latest_from_table_or_query/get_historical_features/pull_all_from_table_or_query→Trino客户端封装执行 SQL →TrinoRetrievalJob暴露to_df/to_arrow/to_trino/persist等消费接口其中类型转换统一走trino_type_map实体表上传走connectors/upload.py。小结Trino 离线存储为 Feast 提供了一条 SQL-on-anything 的离线取数路径通过TrinoOfflineStoreConfig声明集群连接与认证用TrinoSource对接 Trino 表或任意查询由MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN模板保证点对点时间连接的正确性再借TrinoRetrievalJob统一输出 DataFrame / Arrow / 目标表 / SavedDataset。本文覆盖的配置字段、认证方式、SQL 模板语义、类型映射与本地测试命令均可在仓库对应源码文件中逐一验证可直接用于接入 Trino 数据湖或跨源联邦查询的训练数据生产。【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feast创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考