Feast on Amazon 解决方案

{"value":"\n\n#### **背景&引言**\n\n\n众所周知，AI 算法模型开发落地有三个主要阶段：数据准备、模型训练、模型部署。目前已经有较多厂商及开源社区推出通用的 AI、MLOps 平台支撑模型训练与部署阶段，但主要偏重于机器学习模型开发，部署，服务层面，自 2019 年后才陆续有各厂商推出数据准备支撑阶段的产品及服务，即特征平台(如 [Amazon Sagemaker feature Store](https://docs.aws.amazon.com/zh_cn/sagemaker/latest/dg/feature-store.html))。\n\n特征平台的主要能力包含：特征注册中心、离线存储&消费、在线存储&消费、离线&在线特征同步，特征版本，尤其特征版本最为重要，实现特征 point-in-time cross join，避免特征穿越造成 train-server skew 的重要功能特性。\n\n各个厂商在特征平台的架构和实现方式方面迥然不同，缺乏跨平台的通用的特征库方案。\n\nFeast (**Fea**ture **St**ore) 是一套开源特征库框架，纯 python 框架，与 Pandas dataframe 无缝集成，对 ML，AI 算法工程师友好，它提供了在线，离线特征库注册，特征库存储，特征数据摄取、训练数据检索、特征版本、离线-在线特征同步等功能；且具有云原生亲和力，可以构建在多个公有云平台上。\n\n本文介绍了 Feast 框架的整体架构及设计思路，并 step by step 详细说明了 Feast on Amazon 集成和使用，包括安装部署离线/在线特征库、使用特征库、特征库同步的方法等。对于使用 Feast 开源框架构建 MLOps 平台的用户，本文可以作为快速构建和开发指南。\n\n\n#### **Feast 整体架构**\n\n\n![image.png](https://dev-media.amazoncloud.cn/e9c40747242947c4818e05e014ee05de_image.png)\n\nFeast的主要功能组件：\n\n- **Feast Repo&Registry**:轻量级的目录级及 Split 文件数据库格式 Repository，用于特征库基础设施及元数据注册\n- **Feast Python SDK/CLI**: 开发构建及使用特征库的主要功能组件\n - **Feast Apply**:命令行工具执行安装部署配置的特征库到底层基础设施，并且注册特征库元数据到 Runtime 运行态\n - **Feast Materialize**:离线-在线特征库版本同步工具\n - **Get Online Features**:在线特征数据提取，调用对应的在线特征库基础设施 API 抽取特征数据，用于模型推断\n - **Get Historical Features**:离线特征数据抽取，调用对应的离线特征库基础设施 API 抽取历史特征数据，用于模型训练或者特征组合\n- **Online Store**: 在线特征库，根据不同云厂商的 nosql 数据库承载，存储特征快照版本数据\n- **Offline Store**:离线特征库，根据不同云厂商数仓承载，存储特征历史版本数据\n\n\n#### **Feast On Amazon 安装部署方案**\n\n\n##### **依赖准备**\n\n\n- Feast on Amazon 使用 Redshift 作为离线特征库，需要 Redshift 集群(如果采用 Spectrum 外部表，还需要 Spectrum 角色及 Glue Catalog 权限）\n- Feast on Amazon 使用 DynamoDB 作为在线特征库，需要 DynamoDB 读写权限\n- 可以用 Terraform 或者 CloudFormation 准备需要的 Redshift，DDB，IAM 角色等\n- 以下使用 Terraform 为例安装部署 Feast 需要的 Redshift，S3，IAM 角色等各种基础设施\n\n1) 安装部署 Terraform\n\n```\nsudo yum install python3-devel\nsudo yum install -y yum-utils\nsudo yum-config-manager —add-repo https://rpm.releases.hashicorp.com/AmazonLinux/hashicorp.repo\nsudo yum -y install terraform\n```\n\n2) 编写 Terraform 配置文件\n\n```\nproject: feast_aws_repo\nregistry: data/registry.db\nprovider: aws\nonline_store:\n type: dynamodb\n region: ap-southeast-1\noffline_store:\n type: redshift\n cluster_id: feast-demo2-redshift-cluster\n region: ap-southeast-1\n database: flinkstreamdb\n user: awsuser\n s3_staging_location: s3://feastdemobucket\n iam_role: arn:aws:iam::**********:role/s3_spectrum_role\n```\n\n3) 构建基础设施\n\n```\ncd infra\nsudo terraform init\nsudo terraform plan -var=\"admin_password=xxxxx\"\nsudo terraform apply -var=\"admin_password=xxxxx\"\n```\n\n4) 如果需要 Spectrum 承载离线特征库，需要在 Redshift 中建立 Spectrum 外部 schema，以便指向Glue Catalog 中的 s3 外部表\n\n```\naws redshift-data execute-statement \\\n —region ap-southeast-1 \\\n —cluster-identifier feast-demo-redshift-cluster \\\n —db-user awsuser \\\n —database dev —sql \"create external schema spectrum from data catalog database 'flinkstreamdb' iam_role \\\n 'arn:aws:iam::**********:role/s3_spectrum_role' create external database if not exists;“\n```\n\n\n##### **Feast 特征库 Repository 准备**\n\n\n1) 依赖安装及升级\n\n```\npip3 install -U numpy==1.21\npip3 install feast[aws]\n```\n\n2) 初始化 repository\n\n```\nfeast init -t xxxxx（repository_name）\nAWS Region (e.g. us-west-2): ap-southeast-1\nRedshift Cluster ID: feast-demo-redshift-cluster\nRedshift Database Name: flinkstreamdb\nRedshift User Name: awsuser\nRedshift S3 Staging Location (s3://*): s3://feastdemobucket\nRedshift IAM Role for S3 (arn:aws:iam::*:role/*): arn:aws:iam::xxxxxx:role/s3_spectrum_role\n```\n\n创建好的特征库的 schema 及骨架示例：\n\n```\n$ tree ./feast_aws_repo/\n./feast_aws_repo/\n├── data\n│ └── registry.db\n├── driver_repo.py\n├── feature_store.yaml\n```\n\n- *.yam l 配置指定 Feast repository 的基础环境资源（s3、Redshift、DDB 等)\n- *.py 配置特征库元数据，特征 view 及 schema 等\n- db 保存基于 *.py 元数据构建后的特征组，特征库对象实例，以便运行态使用\n\n安装部署后的 feature_store.yaml 示例：\n\n```\nproject: feast_aws_repo\nregistry: data/registry.db\nprovider: aws\nonline_store:\n type: dynamodb\n region: ap-southeast-1\noffline_store:\n type: redshift\n cluster_id: feast-demo2-redshift-cluster\n region: ap-southeast-1\n database: flinkstreamdb\n user: awsuser\n s3_staging_location: s3://feastdemobucket\n iam_role: arn:aws:iam::xxxxxxx:role/s3_spectrum_role\n```\n\ndriver_repo 的司机行程特征库元数据示例：\n\n```\nfrom datetime import timedelta\nfrom feast import Entity, Feature, FeatureView, RedshiftSource, ValueType\ndriver = Entity(\n name=\"driver_id\",\n join_key=\"driver_id\",\n value_type=ValueType.INT64,\n)\ndriver_stats_source = RedshiftSource(\n table=\"feast_driver_hourly_stats\",\n event_timestamp_column=\"event_timestamp\",\n created_timestamp_column=\"created\",\n)\n\ndriver_stats_fv = FeatureView(\n name=\"driver_hourly_stats\",\n entities=[\"driver_id\"],\n ttl=timedelta(weeks=52),\n features=[\n Feature(name=\"conv_rate\", dtype=ValueType.FLOAT),\n Feature(name=\"acc_rate\", dtype=ValueType.FLOAT),\n Feature(name=\"avg_daily_trips\", dtype=ValueType.INT64),\n ],\n batch_source=driver_stats_source,\n tags={\"team\": \"driver_performance\"},\n)\n```\n\n部署成功后可以在 Redshift 看到离线特征库的 Spectuam schema 及库表，DDB 中可以看到在线特征库的表\n\nRedshift 离线特征库：\n\n![image.png](https://dev-media.amazoncloud.cn/09ed41c597d64ff38457a91c27ba0fa5_image.png)\n\nDDB 在线特征库：\n\n![image.png](https://dev-media.amazoncloud.cn/e4b743025fbc49d0a2204d362c478a31_image.png)\n\n##### **使用Feast SDK API进行特征库操作**\n\n\n###### **连接特征库**\n\n\n安装部署完成后，在 python 代码中，可以方便的通过加载注册的 repository 路径，来连接到特征库及特征组\n\n在 repository 中注册的特征组，也可以直接 import 实例化\n\n```\nfrom datetime import datetime, timedelta\nimport pandas as pd\nfrom feast import FeatureStore\nfrom driver_repo import driver, driver_stats_fv\nfs = FeatureStore(repo_path=\"./\")\n>>> print(fs)\n<feast.feature_store.FeatureStore object at 0x7f48d47098d0>\n>>> print(driver_stats_fv)\n{\n \"spec\": {\n \"name\": \"driver_hourly_stats\",\n \"entities\": [\n \"driver_id\"\n ],\n \"features\": [\n {\n \"name\": \"conv_rate\",\n \"valueType\": \"FLOAT\"\n },\n {\n \"name\": \"acc_rate\",\n \"valueType\": \"FLOAT\"\n },\n {\n \"name\": \"avg_daily_trips\",\n \"valueType\": \"INT64\"\n }\n ],\n \"tags\": {\n \"team\": \"driver_performance\"\n },\n \"ttl\": \"31449600s\",\n \"batchSource\": {\n \"type\": \"BATCH_REDSHIFT\",\n \"eventTimestampColumn\": \"event_timestamp\",\n \"createdTimestampColumn\": \"created\",\n \"redshiftOptions\": {\n \"table\": \"feast_driver_hourly_stats\"\n },\n \"dataSourceClassType\": \"feast.infra.offline_stores.redshift_source.RedshiftSource\"\n },\n \"online\": true\n },\n \"meta\": {}\n}\n```\n\n\n###### **离线特征数据提取**\n\n\n通过 Feast get_historical_features API，可以抽取离线特征库数据用于离线训练或特征组合\n\n```\nfeatures = [\"driver_hourly_stats:conv_rate\", \"driver_hourly_stats:acc_rate\"]\nentity_df = pd.DataFrame(\n {\n \"event_timestamp\": [\n pd.Timestamp(dt, unit=\"ms\", tz=\"UTC\").round(\"ms\")\n for dt in pd.date_range(\n start=datetime.now() - timedelta(days=3),\n end=datetime.now(),\n periods=3,\n )\n ],\n \"driver_id\": [1001, 1002, 1003],\n }\n )\n training_df = fs.get_historical_features(\n features=features, entity_df=entity_df\n ).to_df()\n```\n\n如上我们抽取特征标识（entity 字段为 driver_id）为 1001,1002,1003, 时间版本为最近 3 天的离线特征库数据\n\n```\n>>> training_df\n event_timestamp driver_id conv_rate acc_rate\n0 2022-07-04 02:33:54.114 1001 0.036082 0.707744\n1 2022-07-05 14:33:54.114 1002 0.522306 0.983233\n2 2022-07-07 02:33:54.114 1003 0.734294 0.034062\n```\n\n\n###### **离线特征组合**\n\n\n多个特征组需要联合并抽取作为模型训练时，get_historical_features 可以指定多个特征 view 的 features，基于 event_timestamp 做 point-in-time 关联，从而得到同一时间版本的离线特征组合的数据\n\n ```\nfeast_features = [\n \"zipcode_features:city\",\n \"zipcode_features:state\",\n \"zipcode_features:location_type\",\n \"zipcode_features:tax_returns_filed\",\n \"zipcode_features:population\",\n \"zipcode_features:total_wages\",\n \"credit_history:credit_card_due\",\n \"credit_history:mortgage_due\",\n \"credit_history:student_loan_due\",\n \"credit_history:vehicle_loan_due\",\n \"credit_history:hard_pulls\",\n \"credit_history:missed_payments_2y\",\n \"credit_history:missed_payments_1y\",\n \"credit_history:missed_payments_6m\",\n \"credit_history:bankruptcies\",\n ]\ntraining_df = self.fs.get_historical_features(\n entity_df=entity_df, features=feast_features\n).to_df()\n```\n\n如上代码示例，在抽取离线特征时，关联了 credit_history 和 zipcode_features 两个离线特征库的相应特征字段，Feast 会在后台拼接 Redshift Sql 关联对应的库表及 event_timestamp 等条件\n\n\n###### **离线特征数据同步在线特征库**\n\n\n通过Feast 提供的 ```materialize cli```，可以将指定时间版本的 Redshift 离线特征数据同步到 DynamoDB 的在线特征库中\n\nmaterialize-incremental cli 会记录该 repository 特征库下每次同步的增量时间版本，因此每次执行会把自上次执行至今的最新数据增量同步到 DynamoDB\n\n```\nCURRENT_TIME=$(date -u +\"%Y-%m-%dT%H:%M:%S\")\nfeast materialize-incremental $CURRENT_TIME\n\nMaterializing 1 feature views to 2022-07-07 08:00:03+00:00 into the sqlite online\nstore.\ndriver_hourly_stats from 2022-07-06 16:25:47+00:00 to 2022-07-07 08:00:03+00:00:\n100%|████████████████████████████████████████████| 5/5 [00:00<00:00, 592.05it/s]\n```\n\n当然也可以使用 ```materialize``` 显式指定开始时间（startdt）和截止时间（enddt）, feast 会将指定时间版本的离线特征库数据同步到在线特征库\n\n```\nfeast materialize 2022-07-13T00:00:00 2022-07-19T00:00:00\n\nMaterializing 1 feature views from 2022-07-13 00:00:00+00:00 to 2022-07-19 00:00:00+00:00 into the dynamodb online store.\ndriver_hourly_stats:\n100%|█████████████████████████████████████████████████████████████████| 5/5 [00:00<00:00, 51.18it/s]\n```\n\n\n###### **在线特征查询**\n\n\n```\n>>> online_features = fs.get_online_features(\n features=features, entity_rows=[{\"driver_id\": 1001}, {\"driver_id\": 1002}],\n ).to_dict()\n>>> print(pd.DataFrame.from_dict(online_features))\n acc_rate conv_rate driver_id\n0 0.179407 0.984951 1001\n1 0.023422 0.069323 1002\n```\n\n\n#### **Feast offline store on Spark 方案**\n\n\n上文我们看到的是 Feast 依托 Amazon Redshift 作为离线特征库存储和特征抽取的方案，虽然安装部署简介明快，上手方便，但 Redshift 定位是云服务数据仓库，虽然在 sql 兼容性、扩展性上优秀，但灵活性不足，如：\n\n- 离线特征抽取必须要指定 event_timestamp 版本，无法直接查询最新 snapshot\n- point-in-time 关联查询直接拼接 partition over 分组 sql 并下压，海量数据情况下，多历史版本的特征库 time travel 抽取时会膨胀数倍，存在性能瓶颈\n\nFeast 自0.19版本开始，支持 Spark 作为离线特征库历史数据提取，版本查询，同步在线特征库的计算框架\n\nSpark 作为高性能分布式计算引擎，在海量数据场景下性能优异，且使用 Spark 时，Feast FeatureView 的 DataSource 既可以是指向 Hive 中的表，也可以是指向对象存储上的文件，通过 Hive 表可以兼容诸如 Hudi、iceberg 等多种数据湖架构。\n\n同时，通过 Spark 离线特征库抽取的特征数据，Feast 将其封装为 Spark DataFrame，从而可以方便的加载到 S3分布式存储，因而也避免了 Pandas DataFrame 保存在本地磁盘的存储空间问题。\n\n\n##### **Feast point-in-time correct join Spark 实现**\n\n\npoint-in-time correct join，根据源码来看，使用 pySpark+SparkSQL 实现，因此整体思路和 Redshift 类似：\n\n- 将 entity_df 由 DataFrame 转化为 Spark DataFrame，并注册成临时表\n- 根据用户指定要关联的 features，找到对应的 FeatureView，进而找到底层的 DataSource 和相关的元数据\n- 根据以上信息，即 query_context，通过 jinjia 渲染一个 SparkSQL，并提交给 Spark 集群计算\n- 计算完成的结果就是实现 point-in-time correct join 之后的 training dataset\n\n\n##### **Feast offline store on Amazon EMR安装部署**\n\n\nAmazon EMR 是全托管的 hadoop 大数据集群，提供了良好的弹性伸缩，高可用，存算分离等特性，且通过 EMRFS 原生集成 Amazon S3云存储，用于承载 Feast 的 Spark 离线特征库具有天然的亲和力。\n\n以下详细介绍 Feast Spark 离线特征库在 Amazon EMR 的安装部署步骤及使用方法\n\n\n###### **启动 Amazon EMR 集群**\n\n\nAmazon EMR 的启动方法本文不再赘述，感兴趣的同学可以参阅 Amazon EMR 文档\n\n此处选择 emr 6.5版本，Spark 3.1.2\n\n![image.png](https://dev-media.amazoncloud.cn/014d764b306f4627b4ee11e65c521e7a_image.png)\n\n\n###### **Offline store on EMR 特征库配置**\n\n\n我们在 emr 主节点上可以 feast init 特征库，从而直接利用 Amazon EMR上spark 与 S3的原生集成，通过 emrfs 读写 S3数据湖上各种格式文件，不再需要 hadoop s3开源 lib 的支持\n\nfeast init my_project 后，在该特征库的 yaml 配置文件中，指定 Feast spark 的对应参数即可：\n\n```\nproject: feast_spark_project\nregistry: data/registry.db\nprovider: local\noffline_store:\n type: spark\n spark_conf:\n spark.master: yarn\n spark.ui.enabled: \"true\"\n spark.eventLog.enabled: \"true\"\n spark.sql.catalogImplementation: \"hive\"\n spark.sql.parser.quotedRegexColumnNames: \"true\"\n spark.sql.session.timeZone: \"UTC\"\n```\n\n配置完成后，通过 feast apply cli 同样部署到 EMR spark\n\n注：在 EMR master 节点上 pyspark lib 路径需要在环境变量中设置，以便 feast 找到 spark 的 home 目录及相应配置\n\n```\nsource /etc/spark/conf/spark-env.sh\nexport PYTHONPATH=\"${SPARK_HOME}/python/:$PYTHONPATH\"\nexport PYTHONPATH=\"${SPARK_HOME}/python/lib/py4j-0.10.9-src.zip:$PYTHONPATH\"\n```\n\n###### **Feast on Spark 离线特征库元数据**\n\n```\nfrom feast.infra.offline_stores.contrib.spark_offline_store.spark_source import ( SparkSource,)\n\ndriver_hourly_stats= SparkSource(\n name=\"driver_hourly_stats\",\n query=\"SELECT event_timestamp as ts, created_timestamp as created, conv_rate,conv_rate,conv_rate FROM emr_feature_store.driver_hourly_stats\",\n event_timestamp_column=\"ts\",\n created_timestamp_column=\"created\"\n )\n\n```\n \nFeast 的 sparkSource 提供了 query， table，及原始 raw 文件路径几种初始化方法，本文中使用 query 方式。\n\n需要注意 query 方式中，需要指定 event timestamp field 特征字段以便 Feast 识别作为 point-in-time cross join 时间版本抽取及特征 join 的依据\n\n\n###### **Feast Spark offline store 执行**\n\n\n配置 Spark 作为 Feast offline store 后，通过 Amazon EMR 上 spark history UI，可以清楚的看到其 get_historical_features 方法，底层 Feast 使用 SparkSQL 创建临时视图，拼接 event time join 的 sql，并查询上文中 source 数据湖上 hive 库表等各个步骤的业务逻辑：\n\n![image.png](https://dev-media.amazoncloud.cn/116184a35b2741b8a940beb38d37c77b_image.png)\n\n跟踪 Spark history UI 上，Spark Sql 的各个 query 可以看到，Feast 的 get_historical_features 方法执行时，会构造临时表 entity_dataframe，即用户调用 get_historical_features 方法时，传入的样本列表。再构建 driver_hourly_stats_base，即需要 join 及 point-in-time 查询的即样例特征时序表\n\n```\n== Parsed Logical Plan ==\n'CreateViewStatement [driver_hourly_stats__cleaned], (\n\n WITH driver_hourly_stats__entity_dataframe AS (\n SELECT\n driver_id,\n entity_timestamp,\n driver_hourly_stats__entity_row_unique_id\n FROM entity_dataframe\n GROUP BY\n driver_id,\n entity_timestamp,\n driver_hourly_stats__entity_row_unique_id\n ),\n\ndriver_hourly_stats__base AS (\n SELECT\n subquery.*,\n entity_dataframe.entity_timestamp,\n entity_dataframe.driver_hourly_stats__entity_row_unique_id\n FROM driver_hourly_stats__subquery AS subquery\n INNER JOIN driver_hourly_stats__entity_dataframe AS entity_dataframe\n ON TRUE\n AND subquery.event_timestamp <= entity_dataframe.entity_timestamp\n\n \n AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - 86400 * interval '1' second\n \n\n \n AND subquery.driver_id = entity_dataframe.driver_id\n \n ),\n```\n\n后续的 subquery、dedup 及 cleaned 子查询，会基于以上的两张基础表，进行基于特征标识字段 driver_id 和时序时间戳字段 event_timestamp 的分组排序，剔重等操作，最后 join 样本列表临时表 entity_dataframe，整个流程与 Redshift 上基本一致\n\n```\ndriver_hourly_stats__subquery AS (\n SELECT\n ts as event_timestamp,\n created as created_timestamp,\n driver_id AS driver_id,\n \n conv_rate as conv_rate, \n \n acc_rate as acc_rate\n \n FROM (SELECT driver_id,event_timestamp as ts, created_timestamp as created, conv_rate,acc_rate,avg_daily_trips FROM emr_feature_store.driver_hourly_stats)\n WHERE ts <= '2022-07-25T03:27:05.903000'\n \n AND ts >= '2022-07-21T03:27:05.903000'\n \n ),\n\n driver_hourly_stats__dedup AS (\n SELECT\n driver_hourly_stats__entity_row_unique_id,\n event_timestamp,\n MAX(created_timestamp) as created_timestamp\n FROM driver_hourly_stats__base\n GROUP BY driver_hourly_stats__entity_row_unique_id, event_timestamp\n ),\ndriver_hourly_stats__latest AS (\n SELECT\n event_timestamp,\n created_timestamp,\n driver_hourly_stats__entity_row_unique_id\n FROM\n (\n SELECT *,\n ROW_NUMBER() OVER(\n PARTITION BY driver_hourly_stats__entity_row_unique_id\n ORDER BY event_timestamp DESC,created_timestamp DESC\n ) AS row_number\n FROM driver_hourly_stats__base\n \n INNER JOIN driver_hourly_stats__dedup\n USING (driver_hourly_stats__entity_row_unique_id, event_timestamp, created_timestamp)\n \n )\n WHERE row_number = 1\n )\n```\n\nAPI 结果返回可以 to_df 为 Spark 的 Dataframe，从而实现 remote 存储离线特征库抽取结果数据的操作，这也从另一方面解决了原有 Redshift 离线特征存储，特征抽取只能返回 pandas Dataframe 的劣势，在大数据量离线特征场景下更有优势\n\n\n#### **总结**\n\n\n综上所述，Feast 框架整体架构和在 Amazon 的构建是非常简洁明快的，对构建 MLOps 平台的用户而言，其主要有价值的优势如下：\n\n- 同时提供了离线，在线特征库，离线-在线特征库快照版本同步功能\n- 轻量级，快速部署使用，代码即配置，feast apply 即可部署到 Amazon\n- 通过 repository 文件系统隔离特征库，方便 MLOps 多租户多 CICD 协同开发\n- API 抽象程度高，贴近 AI/ML 算法工程师业务语言\n\n对于海量离线特征数据抽取时 point-in-time cross join 的版本查询数据膨胀的业界难点，Feast 也可以通过 on EMR Spark 的构建方式，优化解决其性能问题\n\n\n#### **参考资料**\n\n\nAmazon Sagemaker Feature Store: [https://docs.aws.amazon.com/zh_cn/sagemaker/latest/dg/feature-store.html](https://docs.aws.amazon.com/zh_cn/sagemaker/latest/dg/feature-store.html)\n\nFeast官方：[https://docs.feast.dev/getting-started/architecture-and-components/overview](https://docs.feast.dev/getting-started/architecture-and-components/overview)\n\nAmazon EMR集群部署：[https://docs.aws.amazon.com/zh_cn/emr/latest/ManagementGuide/emr-what-is-emr.html](https://docs.aws.amazon.com/zh_cn/emr/latest/ManagementGuide/emr-what-is-emr.html)\n\n\n#### **本篇作者**\n\n\n![image.png](https://dev-media.amazoncloud.cn/251089bc94a3449988ff3e039457bd75_image.png)\n\n\n#### **唐清原**\n\n\nAmazon 数据分析解决方案架构师，负责 Amazon Data Analytic 服务方案架构设计以及性能优化，迁移，治理等 Deep Dive 支持。10+数据领域研发及架构设计经验，历任 Oracle 高级咨询顾问，咪咕文化数据集市高级架构师，澳新银行数据分析领域架构师职务。在大数据，数据湖，智能湖仓，及相关推荐系统 /MLOps 平台等项目有丰富实战经验","render":"<h4><a id=\"_2\"></a><strong>背景&引言</strong></h4>\n<p>众所周知，AI 算法模型开发落地有三个主要阶段：数据准备、模型训练、模型部署。目前已经有较多厂商及开源社区推出通用的 AI、MLOps 平台支撑模型训练与部署阶段，但主要偏重于机器学习模型开发，部署，服务层面，自 2019 年后才陆续有各厂商推出数据准备支撑阶段的产品及服务，即特征平台(如 <a href=\"https://docs.aws.amazon.com/zh_cn/sagemaker/latest/dg/feature-store.html\" target=\"_blank\">Amazon Sagemaker feature Store</a>)。</p>\n<p>特征平台的主要能力包含：特征注册中心、离线存储&消费、在线存储&消费、离线&在线特征同步，特征版本，尤其特征版本最为重要，实现特征 point-in-time cross join，避免特征穿越造成 train-server skew 的重要功能特性。</p>\n<p>各个厂商在特征平台的架构和实现方式方面迥然不同，缺乏跨平台的通用的特征库方案。</p>\n<p>Feast (<strong>Fea</strong>ture <strong>St</strong>ore) 是一套开源特征库框架，纯 python 框架，与 Pandas dataframe 无缝集成，对 ML，AI 算法工程师友好，它提供了在线，离线特征库注册，特征库存储，特征数据摄取、训练数据检索、特征版本、离线-在线特征同步等功能；且具有云原生亲和力，可以构建在多个公有云平台上。</p>\n<p>本文介绍了 Feast 框架的整体架构及设计思路，并 step by step 详细说明了 Feast on Amazon 集成和使用，包括安装部署离线/在线特征库、使用特征库、特征库同步的方法等。对于使用 Feast 开源框架构建 MLOps 平台的用户，本文可以作为快速构建和开发指南。</p>\n<h4><a id=\"Feast__16\"></a><strong>Feast 整体架构</strong></h4>\n<p><img src=\"https://dev-media.amazoncloud.cn/e9c40747242947c4818e05e014ee05de_image.png\" alt=\"image.png\" /></p>\n<p>Feast的主要功能组件：</p>\n<ul>\n<li><strong>Feast Repo&Registry</strong>:轻量级的目录级及 Split 文件数据库格式 Repository，用于特征库基础设施及元数据注册</li>\n<li><strong>Feast Python SDK/CLI</strong>: 开发构建及使用特征库的主要功能组件\n<ul>\n<li><strong>Feast Apply</strong>:命令行工具执行安装部署配置的特征库到底层基础设施，并且注册特征库元数据到 Runtime 运行态</li>\n<li><strong>Feast Materialize</strong>:离线-在线特征库版本同步工具</li>\n<li><strong>Get Online Features</strong>:在线特征数据提取，调用对应的在线特征库基础设施 API 抽取特征数据，用于模型推断</li>\n<li><strong>Get Historical Features</strong>:离线特征数据抽取，调用对应的离线特征库基础设施 API 抽取历史特征数据，用于模型训练或者特征组合</li>\n</ul>\n</li>\n<li><strong>Online Store</strong>: 在线特征库，根据不同云厂商的 nosql 数据库承载，存储特征快照版本数据</li>\n<li><strong>Offline Store</strong>:离线特征库，根据不同云厂商数仓承载，存储特征历史版本数据</li>\n</ul>\n<h4><a id=\"Feast_On_Amazon__33\"></a><strong>Feast On Amazon 安装部署方案</strong></h4>\n<h5><a id=\"_36\"></a><strong>依赖准备</strong></h5>\n<ul>\n<li>Feast on Amazon 使用 Redshift 作为离线特征库，需要 Redshift 集群(如果采用 Spectrum 外部表，还需要 Spectrum 角色及 Glue Catalog 权限）</li>\n<li>Feast on Amazon 使用 DynamoDB 作为在线特征库，需要 DynamoDB 读写权限</li>\n<li>可以用 Terraform 或者 CloudFormation 准备需要的 Redshift，DDB，IAM 角色等</li>\n<li>以下使用 Terraform 为例安装部署 Feast 需要的 Redshift，S3，IAM 角色等各种基础设施</li>\n</ul>\n<ol>\n<li>安装部署 Terraform</li>\n</ol>\n<pre><code class=\"lang-\">sudo yum install python3-devel\nsudo yum install -y yum-utils\nsudo yum-config-manager —add-repo https://rpm.releases.hashicorp.com/AmazonLinux/hashicorp.repo\nsudo yum -y install terraform\n</code></pre>\n<ol start=\"2\">\n<li>编写 Terraform 配置文件</li>\n</ol>\n<pre><code class=\"lang-\">project: feast_aws_repo\nregistry: data/registry.db\nprovider: aws\nonline_store:\n type: dynamodb\n region: ap-southeast-1\noffline_store:\n type: redshift\n cluster_id: feast-demo2-redshift-cluster\n region: ap-southeast-1\n database: flinkstreamdb\n user: awsuser\n s3_staging_location: s3://feastdemobucket\n iam_role: arn:aws:iam::**********:role/s3_spectrum_role\n</code></pre>\n<ol start=\"3\">\n<li>构建基础设施</li>\n</ol>\n<pre><code class=\"lang-\">cd infra\nsudo terraform init\nsudo terraform plan -var="admin_password=xxxxx"\nsudo terraform apply -var="admin_password=xxxxx"\n</code></pre>\n<ol start=\"4\">\n<li>如果需要 Spectrum 承载离线特征库，需要在 Redshift 中建立 Spectrum 外部 schema，以便指向Glue Catalog 中的 s3 外部表</li>\n</ol>\n<pre><code class=\"lang-\">aws redshift-data execute-statement \\\n —region ap-southeast-1 \\\n —cluster-identifier feast-demo-redshift-cluster \\\n —db-user awsuser \\\n —database dev —sql "create external schema spectrum from data catalog database 'flinkstreamdb' iam_role \\\n 'arn:aws:iam::**********:role/s3_spectrum_role' create external database if not exists;“\n</code></pre>\n<h5><a id=\"Feast__Repository__93\"></a><strong>Feast 特征库 Repository 准备</strong></h5>\n<ol>\n<li>依赖安装及升级</li>\n</ol>\n<pre><code class=\"lang-\">pip3 install -U numpy==1.21\npip3 install feast[aws]\n</code></pre>\n<ol start=\"2\">\n<li>初始化 repository</li>\n</ol>\n<pre><code class=\"lang-\">feast init -t xxxxx（repository_name）\nAWS Region (e.g. us-west-2): ap-southeast-1\nRedshift Cluster ID: feast-demo-redshift-cluster\nRedshift Database Name: flinkstreamdb\nRedshift User Name: awsuser\nRedshift S3 Staging Location (s3://*): s3://feastdemobucket\nRedshift IAM Role for S3 (arn:aws:iam::*:role/*): arn:aws:iam::xxxxxx:role/s3_spectrum_role\n</code></pre>\n<p>创建好的特征库的 schema 及骨架示例：</p>\n<pre><code class=\"lang-\">$ tree ./feast_aws_repo/\n./feast_aws_repo/\n├── data\n│ └── registry.db\n├── driver_repo.py\n├── feature_store.yaml\n</code></pre>\n<ul>\n<li>*.yam l 配置指定 Feast repository 的基础环境资源（s3、Redshift、DDB 等)</li>\n<li>*.py 配置特征库元数据，特征 view 及 schema 等</li>\n<li>db 保存基于 *.py 元数据构建后的特征组，特征库对象实例，以便运行态使用</li>\n</ul>\n<p>安装部署后的 feature_store.yaml 示例：</p>\n<pre><code class=\"lang-\">project: feast_aws_repo\nregistry: data/registry.db\nprovider: aws\nonline_store:\n type: dynamodb\n region: ap-southeast-1\noffline_store:\n type: redshift\n cluster_id: feast-demo2-redshift-cluster\n region: ap-southeast-1\n database: flinkstreamdb\n user: awsuser\n s3_staging_location: s3://feastdemobucket\n iam_role: arn:aws:iam::xxxxxxx:role/s3_spectrum_role\n</code></pre>\n<p>driver_repo 的司机行程特征库元数据示例：</p>\n<pre><code class=\"lang-\">from datetime import timedelta\nfrom feast import Entity, Feature, FeatureView, RedshiftSource, ValueType\ndriver = Entity(\n name="driver_id",\n join_key="driver_id",\n value_type=ValueType.INT64,\n)\ndriver_stats_source = RedshiftSource(\n table="feast_driver_hourly_stats",\n event_timestamp_column="event_timestamp",\n created_timestamp_column="created",\n)\n\ndriver_stats_fv = FeatureView(\n name="driver_hourly_stats",\n entities=["driver_id"],\n ttl=timedelta(weeks=52),\n features=[\n Feature(name="conv_rate", dtype=ValueType.FLOAT),\n Feature(name="acc_rate", dtype=ValueType.FLOAT),\n Feature(name="avg_daily_trips", dtype=ValueType.INT64),\n ],\n batch_source=driver_stats_source,\n tags={"team": "driver_performance"},\n)\n</code></pre>\n<p>部署成功后可以在 Redshift 看到离线特征库的 Spectuam schema 及库表，DDB 中可以看到在线特征库的表</p>\n<p>Redshift 离线特征库：</p>\n<p><img src=\"https://dev-media.amazoncloud.cn/09ed41c597d64ff38457a91c27ba0fa5_image.png\" alt=\"image.png\" /></p>\n<p>DDB 在线特征库：</p>\n<p><img src=\"https://dev-media.amazoncloud.cn/e4b743025fbc49d0a2204d362c478a31_image.png\" alt=\"image.png\" /></p>\n<h5><a id=\"Feast_SDK_API_189\"></a><strong>使用Feast SDK API进行特征库操作</strong></h5>\n<h6><a id=\"_192\"></a><strong>连接特征库</strong></h6>\n<p>安装部署完成后，在 python 代码中，可以方便的通过加载注册的 repository 路径，来连接到特征库及特征组</p>\n<p>在 repository 中注册的特征组，也可以直接 import 实例化</p>\n<pre><code class=\"lang-\">from datetime import datetime, timedelta\nimport pandas as pd\nfrom feast import FeatureStore\nfrom driver_repo import driver, driver_stats_fv\nfs = FeatureStore(repo_path="./")\n>>> print(fs)\n<feast.feature_store.FeatureStore object at 0x7f48d47098d0>\n>>> print(driver_stats_fv)\n{\n "spec": {\n "name": "driver_hourly_stats",\n "entities": [\n "driver_id"\n ],\n "features": [\n {\n "name": "conv_rate",\n "valueType": "FLOAT"\n },\n {\n "name": "acc_rate",\n "valueType": "FLOAT"\n },\n {\n "name": "avg_daily_trips",\n "valueType": "INT64"\n }\n ],\n "tags": {\n "team": "driver_performance"\n },\n "ttl": "31449600s",\n "batchSource": {\n "type": "BATCH_REDSHIFT",\n "eventTimestampColumn": "event_timestamp",\n "createdTimestampColumn": "created",\n "redshiftOptions": {\n "table": "feast_driver_hourly_stats"\n },\n "dataSourceClassType": "feast.infra.offline_stores.redshift_source.RedshiftSource"\n },\n "online": true\n },\n "meta": {}\n}\n</code></pre>\n<h6><a id=\"_248\"></a><strong>离线特征数据提取</strong></h6>\n<p>通过 Feast get_historical_features API，可以抽取离线特征库数据用于离线训练或特征组合</p>\n<pre><code class=\"lang-\">features = ["driver_hourly_stats:conv_rate", "driver_hourly_stats:acc_rate"]\nentity_df = pd.DataFrame(\n {\n "event_timestamp": [\n pd.Timestamp(dt, unit="ms", tz="UTC").round("ms")\n for dt in pd.date_range(\n start=datetime.now() - timedelta(days=3),\n end=datetime.now(),\n periods=3,\n )\n ],\n "driver_id": [1001, 1002, 1003],\n }\n )\n training_df = fs.get_historical_features(\n features=features, entity_df=entity_df\n ).to_df()\n</code></pre>\n<p>如上我们抽取特征标识（entity 字段为 driver_id）为 1001,1002,1003, 时间版本为最近 3 天的离线特征库数据</p>\n<pre><code class=\"lang-\">>>> training_df\n event_timestamp driver_id conv_rate acc_rate\n0 2022-07-04 02:33:54.114 1001 0.036082 0.707744\n1 2022-07-05 14:33:54.114 1002 0.522306 0.983233\n2 2022-07-07 02:33:54.114 1003 0.734294 0.034062\n</code></pre>\n<h6><a id=\"_284\"></a><strong>离线特征组合</strong></h6>\n<p>多个特征组需要联合并抽取作为模型训练时，get_historical_features 可以指定多个特征 view 的 features，基于 event_timestamp 做 point-in-time 关联，从而得到同一时间版本的离线特征组合的数据</p>\n<pre><code class=\"lang-\">feast_features = [\n "zipcode_features:city",\n "zipcode_features:state",\n "zipcode_features:location_type",\n "zipcode_features:tax_returns_filed",\n "zipcode_features:population",\n "zipcode_features:total_wages",\n "credit_history:credit_card_due",\n "credit_history:mortgage_due",\n "credit_history:student_loan_due",\n "credit_history:vehicle_loan_due",\n "credit_history:hard_pulls",\n "credit_history:missed_payments_2y",\n "credit_history:missed_payments_1y",\n "credit_history:missed_payments_6m",\n "credit_history:bankruptcies",\n ]\ntraining_df = self.fs.get_historical_features(\n entity_df=entity_df, features=feast_features\n).to_df()\n</code></pre>\n<p>如上代码示例，在抽取离线特征时，关联了 credit_history 和 zipcode_features 两个离线特征库的相应特征字段，Feast 会在后台拼接 Redshift Sql 关联对应的库表及 event_timestamp 等条件</p>\n<h6><a id=\"_315\"></a><strong>离线特征数据同步在线特征库</strong></h6>\n<p>通过Feast 提供的 <code>materialize cli</code>，可以将指定时间版本的 Redshift 离线特征数据同步到 DynamoDB 的在线特征库中</p>\n<p>materialize-incremental cli 会记录该 repository 特征库下每次同步的增量时间版本，因此每次执行会把自上次执行至今的最新数据增量同步到 DynamoDB</p>\n<pre><code class=\"lang-\">CURRENT_TIME=$(date -u +"%Y-%m-%dT%H:%M:%S")\nfeast materialize-incremental $CURRENT_TIME\n\nMaterializing 1 feature views to 2022-07-07 08:00:03+00:00 into the sqlite online\nstore.\ndriver_hourly_stats from 2022-07-06 16:25:47+00:00 to 2022-07-07 08:00:03+00:00:\n100%|████████████████████████████████████████████| 5/5 [00:00<00:00, 592.05it/s]\n</code></pre>\n<p>当然也可以使用 <code>materialize</code> 显式指定开始时间（startdt）和截止时间（enddt）, feast 会将指定时间版本的离线特征库数据同步到在线特征库</p>\n<pre><code class=\"lang-\">feast materialize 2022-07-13T00:00:00 2022-07-19T00:00:00\n\nMaterializing 1 feature views from 2022-07-13 00:00:00+00:00 to 2022-07-19 00:00:00+00:00 into the dynamodb online store.\ndriver_hourly_stats:\n100%|█████████████████████████████████████████████████████████████████| 5/5 [00:00<00:00, 51.18it/s]\n</code></pre>\n<h6><a id=\"_343\"></a><strong>在线特征查询</strong></h6>\n<pre><code class=\"lang-\">>>> online_features = fs.get_online_features(\n features=features, entity_rows=[{"driver_id": 1001}, {"driver_id": 1002}],\n ).to_dict()\n>>> print(pd.DataFrame.from_dict(online_features))\n acc_rate conv_rate driver_id\n0 0.179407 0.984951 1001\n1 0.023422 0.069323 1002\n</code></pre>\n<h4><a id=\"Feast_offline_store_on_Spark__357\"></a><strong>Feast offline store on Spark 方案</strong></h4>\n<p>上文我们看到的是 Feast 依托 Amazon Redshift 作为离线特征库存储和特征抽取的方案，虽然安装部署简介明快，上手方便，但 Redshift 定位是云服务数据仓库，虽然在 sql 兼容性、扩展性上优秀，但灵活性不足，如：</p>\n<ul>\n<li>离线特征抽取必须要指定 event_timestamp 版本，无法直接查询最新 snapshot</li>\n<li>point-in-time 关联查询直接拼接 partition over 分组 sql 并下压，海量数据情况下，多历史版本的特征库 time travel 抽取时会膨胀数倍，存在性能瓶颈</li>\n</ul>\n<p>Feast 自0.19版本开始，支持 Spark 作为离线特征库历史数据提取，版本查询，同步在线特征库的计算框架</p>\n<p>Spark 作为高性能分布式计算引擎，在海量数据场景下性能优异，且使用 Spark 时，Feast FeatureView 的 DataSource 既可以是指向 Hive 中的表，也可以是指向对象存储上的文件，通过 Hive 表可以兼容诸如 Hudi、iceberg 等多种数据湖架构。</p>\n<p>同时，通过 Spark 离线特征库抽取的特征数据，Feast 将其封装为 Spark DataFrame，从而可以方便的加载到 S3分布式存储，因而也避免了 Pandas DataFrame 保存在本地磁盘的存储空间问题。</p>\n<h5><a id=\"Feast_pointintime_correct_join_Spark__372\"></a><strong>Feast point-in-time correct join Spark 实现</strong></h5>\n<p>point-in-time correct join，根据源码来看，使用 pySpark+SparkSQL 实现，因此整体思路和 Redshift 类似：</p>\n<ul>\n<li>将 entity_df 由 DataFrame 转化为 Spark DataFrame，并注册成临时表</li>\n<li>根据用户指定要关联的 features，找到对应的 FeatureView，进而找到底层的 DataSource 和相关的元数据</li>\n<li>根据以上信息，即 query_context，通过 jinjia 渲染一个 SparkSQL，并提交给 Spark 集群计算</li>\n<li>计算完成的结果就是实现 point-in-time correct join 之后的 training dataset</li>\n</ul>\n<h5><a id=\"Feast_offline_store_on_Amazon_EMR_383\"></a><strong>Feast offline store on Amazon EMR安装部署</strong></h5>\n<p>Amazon EMR 是全托管的 hadoop 大数据集群，提供了良好的弹性伸缩，高可用，存算分离等特性，且通过 EMRFS 原生集成 Amazon S3云存储，用于承载 Feast 的 Spark 离线特征库具有天然的亲和力。</p>\n<p>以下详细介绍 Feast Spark 离线特征库在 Amazon EMR 的安装部署步骤及使用方法</p>\n<h6><a id=\"_Amazon_EMR__391\"></a><strong>启动 Amazon EMR 集群</strong></h6>\n<p>Amazon EMR 的启动方法本文不再赘述，感兴趣的同学可以参阅 Amazon EMR 文档</p>\n<p>此处选择 emr 6.5版本，Spark 3.1.2</p>\n<p><img src=\"https://dev-media.amazoncloud.cn/014d764b306f4627b4ee11e65c521e7a_image.png\" alt=\"image.png\" /></p>\n<h6><a id=\"Offline_store_on_EMR__401\"></a><strong>Offline store on EMR 特征库配置</strong></h6>\n<p>我们在 emr 主节点上可以 feast init 特征库，从而直接利用 Amazon EMR上spark 与 S3的原生集成，通过 emrfs 读写 S3数据湖上各种格式文件，不再需要 hadoop s3开源 lib 的支持</p>\n<p>feast init my_project 后，在该特征库的 yaml 配置文件中，指定 Feast spark 的对应参数即可：</p>\n<pre><code class=\"lang-\">project: feast_spark_project\nregistry: data/registry.db\nprovider: local\noffline_store:\n type: spark\n spark_conf:\n spark.master: yarn\n spark.ui.enabled: "true"\n spark.eventLog.enabled: "true"\n spark.sql.catalogImplementation: "hive"\n spark.sql.parser.quotedRegexColumnNames: "true"\n spark.sql.session.timeZone: "UTC"\n</code></pre>\n<p>配置完成后，通过 feast apply cli 同样部署到 EMR spark</p>\n<p>注：在 EMR master 节点上 pyspark lib 路径需要在环境变量中设置，以便 feast 找到 spark 的 home 目录及相应配置</p>\n<pre><code class=\"lang-\">source /etc/spark/conf/spark-env.sh\nexport PYTHONPATH="${SPARK_HOME}/python/:$PYTHONPATH"\nexport PYTHONPATH="${SPARK_HOME}/python/lib/py4j-0.10.9-src.zip:$PYTHONPATH"\n</code></pre>\n<h6><a id=\"Feast_on_Spark__433\"></a><strong>Feast on Spark 离线特征库元数据</strong></h6>\n<pre><code class=\"lang-\">from feast.infra.offline_stores.contrib.spark_offline_store.spark_source import ( SparkSource,)\n\ndriver_hourly_stats= SparkSource(\n name="driver_hourly_stats",\n query="SELECT event_timestamp as ts, created_timestamp as created, conv_rate,conv_rate,conv_rate FROM emr_feature_store.driver_hourly_stats",\n event_timestamp_column="ts",\n created_timestamp_column="created"\n )\n\n</code></pre>\n<p>Feast 的 sparkSource 提供了 query， table，及原始 raw 文件路径几种初始化方法，本文中使用 query 方式。</p>\n<p>需要注意 query 方式中，需要指定 event timestamp field 特征字段以便 Feast 识别作为 point-in-time cross join 时间版本抽取及特征 join 的依据</p>\n<h6><a id=\"Feast_Spark_offline_store__452\"></a><strong>Feast Spark offline store 执行</strong></h6>\n<p>配置 Spark 作为 Feast offline store 后，通过 Amazon EMR 上 spark history UI，可以清楚的看到其 get_historical_features 方法，底层 Feast 使用 SparkSQL 创建临时视图，拼接 event time join 的 sql，并查询上文中 source 数据湖上 hive 库表等各个步骤的业务逻辑：</p>\n<p><img src=\"https://dev-media.amazoncloud.cn/116184a35b2741b8a940beb38d37c77b_image.png\" alt=\"image.png\" /></p>\n<p>跟踪 Spark history UI 上，Spark Sql 的各个 query 可以看到，Feast 的 get_historical_features 方法执行时，会构造临时表 entity_dataframe，即用户调用 get_historical_features 方法时，传入的样本列表。再构建 driver_hourly_stats_base，即需要 join 及 point-in-time 查询的即样例特征时序表</p>\n<pre><code class=\"lang-\">== Parsed Logical Plan ==\n'CreateViewStatement [driver_hourly_stats__cleaned], (\n\n WITH driver_hourly_stats__entity_dataframe AS (\n SELECT\n driver_id,\n entity_timestamp,\n driver_hourly_stats__entity_row_unique_id\n FROM entity_dataframe\n GROUP BY\n driver_id,\n entity_timestamp,\n driver_hourly_stats__entity_row_unique_id\n ),\n\ndriver_hourly_stats__base AS (\n SELECT\n subquery.*,\n entity_dataframe.entity_timestamp,\n entity_dataframe.driver_hourly_stats__entity_row_unique_id\n FROM driver_hourly_stats__subquery AS subquery\n INNER JOIN driver_hourly_stats__entity_dataframe AS entity_dataframe\n ON TRUE\n AND subquery.event_timestamp <= entity_dataframe.entity_timestamp\n\n \n AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - 86400 * interval '1' second\n \n\n \n AND subquery.driver_id = entity_dataframe.driver_id\n \n ),\n</code></pre>\n<p>后续的 subquery、dedup 及 cleaned 子查询，会基于以上的两张基础表，进行基于特征标识字段 driver_id 和时序时间戳字段 event_timestamp 的分组排序，剔重等操作，最后 join 样本列表临时表 entity_dataframe，整个流程与 Redshift 上基本一致</p>\n<pre><code class=\"lang-\">driver_hourly_stats__subquery AS (\n SELECT\n ts as event_timestamp,\n created as created_timestamp,\n driver_id AS driver_id,\n \n conv_rate as conv_rate, \n \n acc_rate as acc_rate\n \n FROM (SELECT driver_id,event_timestamp as ts, created_timestamp as created, conv_rate,acc_rate,avg_daily_trips FROM emr_feature_store.driver_hourly_stats)\n WHERE ts <= '2022-07-25T03:27:05.903000'\n \n AND ts >= '2022-07-21T03:27:05.903000'\n \n ),\n\n driver_hourly_stats__dedup AS (\n SELECT\n driver_hourly_stats__entity_row_unique_id,\n event_timestamp,\n MAX(created_timestamp) as created_timestamp\n FROM driver_hourly_stats__base\n GROUP BY driver_hourly_stats__entity_row_unique_id, event_timestamp\n ),\ndriver_hourly_stats__latest AS (\n SELECT\n event_timestamp,\n created_timestamp,\n driver_hourly_stats__entity_row_unique_id\n FROM\n (\n SELECT *,\n ROW_NUMBER() OVER(\n PARTITION BY driver_hourly_stats__entity_row_unique_id\n ORDER BY event_timestamp DESC,created_timestamp DESC\n ) AS row_number\n FROM driver_hourly_stats__base\n \n INNER JOIN driver_hourly_stats__dedup\n USING (driver_hourly_stats__entity_row_unique_id, event_timestamp, created_timestamp)\n \n )\n WHERE row_number = 1\n )\n</code></pre>\n<p>API 结果返回可以 to_df 为 Spark 的 Dataframe，从而实现 remote 存储离线特征库抽取结果数据的操作，这也从另一方面解决了原有 Redshift 离线特征存储，特征抽取只能返回 pandas Dataframe 的劣势，在大数据量离线特征场景下更有优势</p>\n<h4><a id=\"_550\"></a><strong>总结</strong></h4>\n<p>综上所述，Feast 框架整体架构和在 Amazon 的构建是非常简洁明快的，对构建 MLOps 平台的用户而言，其主要有价值的优势如下：</p>\n<ul>\n<li>同时提供了离线，在线特征库，离线-在线特征库快照版本同步功能</li>\n<li>轻量级，快速部署使用，代码即配置，feast apply 即可部署到 Amazon</li>\n<li>通过 repository 文件系统隔离特征库，方便 MLOps 多租户多 CICD 协同开发</li>\n<li>API 抽象程度高，贴近 AI/ML 算法工程师业务语言</li>\n</ul>\n<p>对于海量离线特征数据抽取时 point-in-time cross join 的版本查询数据膨胀的业界难点，Feast 也可以通过 on EMR Spark 的构建方式，优化解决其性能问题</p>\n<h4><a id=\"_563\"></a><strong>参考资料</strong></h4>\n<p>Amazon Sagemaker Feature Store: <a href=\"https://docs.aws.amazon.com/zh_cn/sagemaker/latest/dg/feature-store.html\" target=\"_blank\">https://docs.aws.amazon.com/zh_cn/sagemaker/latest/dg/feature-store.html</a></p>\n<p>Feast官方：<a href=\"https://docs.feast.dev/getting-started/architecture-and-components/overview\" target=\"_blank\">https://docs.feast.dev/getting-started/architecture-and-components/overview</a></p>\n<p>Amazon EMR集群部署：<a href=\"https://docs.aws.amazon.com/zh_cn/emr/latest/ManagementGuide/emr-what-is-emr.html\" target=\"_blank\">https://docs.aws.amazon.com/zh_cn/emr/latest/ManagementGuide/emr-what-is-emr.html</a></p>\n<h4><a id=\"_573\"></a><strong>本篇作者</strong></h4>\n<p><img src=\"https://dev-media.amazoncloud.cn/251089bc94a3449988ff3e039457bd75_image.png\" alt=\"image.png\" /></p>\n<h4><a id=\"_579\"></a><strong>唐清原</strong></h4>\n<p>Amazon 数据分析解决方案架构师，负责 Amazon Data Analytic 服务方案架构设计以及性能优化，迁移，治理等 Deep Dive 支持。10+数据领域研发及架构设计经验，历任 Oracle 高级咨询顾问，咪咕文化数据集市高级架构师，澳新银行数据分析领域架构师职务。在大数据，数据湖，智能湖仓，及相关推荐系统 /MLOps 平台等项目有丰富实战经验</p>\n"}

亚马逊云科技解决方案基于行业客户应用场景及技术领域的解决方案

联系亚马逊云科技专家