一款面向企业的可扩展、统一的数据与AI工程平台

重要链接： Slack & 讨论区。 文档。

什么是Feathr？

Feathr是一个数据与AI工程平台，在LinkedIn已广泛应用于生产环境多年，并于2022年开源。目前，它已成为LF AI & Data基金会下的一个项目。

请阅读我们的相关公告：开源Feathr和Feathr在Azure上，以及LF AI & Data基金会的公告。

Feathr可以帮助您：

• 定义 基于原始数据源（批处理和流式）的数据及特征转换，使用类Python的API。
• 注册 转换规则并为其命名，从而为各种用例获取转换后的数据（特征），包括AI建模、合规性、市场推广等。
• 共享 转换规则及数据（特征）给团队或公司内的其他部门。

Feathr在AI建模中尤为有用，它能够自动计算您的特征转换，并以“时间点正确”的语义将其与训练数据进行关联，避免数据泄露问题；同时支持将特征物化并部署到线上生产环境中使用。

🌟 Feathr亮点

• 原生云集成，架构简化且可扩展。
• 经过6年以上生产环境的严格考验：LinkedIn已持续使用Feathr超过6年，并拥有专门的支持团队。
• 内置优化功能，具备强大的可扩展性：Feathr可处理数十亿行数据及PB级别的海量数据，内置布隆过滤器和加盐连接等优化技术。
• 丰富的转换API，包括基于时间的聚合、滑动窗口连接、查找特征等，所有操作均符合AI所需的“时间点正确”原则。
• 类Python的API和高度可定制的用户自定义函数（UDF），原生支持PySpark和Spark SQL，降低数据科学家的学习曲线。
• 统一的数据转换API适用于离线批处理、流式处理和在线环境。
• Feathr内置的注册中心使命名转换和数据/特征的复用变得轻而易举。

🏃 快速入门Feathr - Feathr沙盒

尝试Feathr最简单的方式是使用Feathr沙盒，这是一个自包含的容器，集成了Feathr的大部分功能，您只需5分钟即可开始高效工作。只需运行以下命令：

# 80: Feathr UI, 8888: Jupyter, 7080: Interpret
docker run -it --rm -p 8888:8888 -p 8081:80 -p 7080:7080 -e GRANT_SUDO=yes feathrfeaturestore/feathr-sandbox:releases-v1.0.0

然后您可以查看Feathr快速入门Jupyter笔记本：

http://localhost:8888/lab/workspaces/auto-w/tree/local_quickstart_notebook.ipynb

运行完笔记本后，所有特征都会在UI中注册，您可以通过以下地址访问Feathr UI：

http://localhost:8081

🛠️ 在本地安装Feathr客户端

如果您想在Python环境中安装Feathr客户端，请使用以下命令：

pip install feathr

或者直接从GitHub获取最新代码：

pip install git+https://github.com/feathr-ai/feathr.git#subdirectory=feathr_project

☁️ 在云端运行Feathr用于生产环境

Feathr与Databricks和Azure Synapse实现了原生集成：

请按照Feathr ARM部署指南在Azure上运行Feathr。该指南提供了使用Azure资源管理器模板进行自动化部署的快速入门方法。

如果您希望手动搭建所有组件，可以参考Feathr CLI部署指南，逐步设置每个资源。

• 请阅读Feathr在Databricks上的快速入门指南，了解如何在Databricks上运行Feathr。
• 请阅读Feathr在Azure Synapse上的快速入门指南，了解如何在Azure Synapse上运行Feathr。

📓 文档

• 欲了解更多关于Feathr的信息，请参阅我们的文档。
• 如需Python API参考，请查阅Python API参考。
• 关于Feathr的技术分享，请参阅此处的幻灯片和此处的幻灯片。相关录像可在这里观看。

🧪 样例

名称	描述	平台
NYC Taxi 演示	快速入门笔记本，展示了如何使用 NYC 出租车费预测样本数据定义、物化和注册特征。	Azure Synapse、Databricks、本地 Spark
Databricks 快速入门 NYC Taxi 演示	包含 NYC 出租车费预测样本数据的 Databricks 快速入门笔记本。	Databricks
特征嵌入	Feathr UDF 示例，展示如何使用预训练的 Transformer 模型和酒店评论样本数据定义并使用特征嵌入。	Databricks
欺诈检测演示	一个示例，演示如何使用用户账户和交易数据等多种数据源来操作 Feature Store。	Azure Synapse、Databricks、本地 Spark
产品推荐演示	带有产品推荐场景的 Feathr Feature Store 示例笔记本	Azure Synapse、Databricks、本地 Spark

🔡 Feathr 主要功能

请阅读 Feathr 全部功能，以获取更多示例。以下是其中几个精选功能：

Feathr UI

Feathr 提供了一个直观的用户界面，使您可以搜索和探索所有可用的特征及其相应的血缘关系。

您可以通过 Feathr UI 搜索特征、识别数据源、跟踪特征血缘关系以及管理访问控制。请查看最新的实时演示 这里，了解 Feathr UI 能为您做些什么。在提示登录时，请使用以下任一账号：

• 工作或学校组织账号，包括 Office 365 订阅用户。
• Microsoft 个人账号，即可以访问 Skype、Outlook.com、OneDrive 和 Xbox LIVE 的账号。

有关 Feathr UI 及其背后注册表的更多信息，请参阅 Feathr 特征注册表。

丰富的 UDF 支持

Feathr 具有高度可定制的 UDF，并与原生 PySpark 和 Spark SQL 集成，从而降低数据科学家的学习曲线：

def add_new_dropoff_and_fare_amount_column(df: DataFrame):
    df = df.withColumn("f_day_of_week", dayofweek("lpep_dropoff_datetime"))
    df = df.withColumn("fare_amount_cents", df.fare_amount.cast('double') * 100)
    return df

batch_source = HdfsSource(name="nycTaxiBatchSource",
                        path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/demo_data/green_tripdata_2020-04.csv",
                        preprocessing=add_new_dropoff_and_fare_amount_column,
                        event_timestamp_column="new_lpep_dropoff_datetime",
                        timestamp_format="yyyy-MM-dd HH:mm:ss")

定义具有时间点正确性的窗口聚合特征

agg_features = [Feature(name="f_location_avg_fare",
                        key=location_id,                          # 特征（组）的查询/连接键
                        feature_type=FLOAT,
                        transform=WindowAggTransformation(        # 窗口聚合变换
                            agg_expr="cast_float(fare_amount)",
                            agg_func="AVG",                       # 在窗口上应用平均聚合
                            window="90d")),                       # 在 90 天窗口内
                ]

agg_anchor = FeatureAnchor(name="aggregationFeatures",
                           source=batch_source,
                           features=agg_features)

在其他特征之上定义派生特征

# 在现有特征之上计算一个新的特征（即派生特征）
derived_feature = DerivedFeature(name="f_trip_time_distance",
                                 feature_type=FLOAT,
                                 key=trip_key,
                                 input_features=[f_trip_distance, f_trip_time_duration],
                                 transform="f_trip_distance * f_trip_time_duration")

# 另一个计算嵌入相似度的例子
user_embedding = Feature(name="user_embedding", feature_type=DENSE_VECTOR, key=user_key)
item_embedding = Feature(name="item_embedding", feature_type=DENSE_VECTOR, key=item_key)

user_item_similarity = DerivedFeature(name="user_item_similarity",
                                      feature_type=FLOAT,
                                      key=[user_key, item_key],
                                      input_features=[user_embedding, item_embedding],
                                      transform="cosine_similarity(user_embedding, item_embedding)")

定义流式特征

有关详细信息，请参阅 流式源摄取指南。

时间点连接

有关详细信息，请参阅 Feathr 中的时间点正确性和时间点连接。

运行 Feathr 示例

按照 快速入门 Jupyter 笔记本 进行尝试。此外，还有一个配套的 快速入门指南，其中对笔记本进行了更详细的解释。

🗣️ 关于 Feathr 的技术分享

• Feathr 简介 - 初学者指南
• 使用 Azure Feature Store (Feathr) 和 SynapseML 进行文档智能
• 笔记本教程：使用 Feathr Feature Store 构建产品推荐机器学习模型
• Feathr 在 Feature Store Summit 上的演讲

⚙️ 云集成与架构

Feathr 组件	云集成
离线存储 – 对象存储	Azure Blob Storage、Azure ADLS Gen2、AWS S3
离线存储 – SQL	Azure SQL DB、Azure Synapse 专用 SQL 池、VM 中的 Azure SQL、Snowflake
流式数据源	Kafka、EventHub
在线存储	Redis、Azure Cosmos DB
特征注册与治理	Azure Purview、ANSI SQL（如 Azure SQL Server）
计算引擎	Azure Synapse Spark 池、Databricks
机器学习平台	Azure 机器学习、Jupyter Notebook、Databricks Notebook
文件格式	Parquet、ORC、Avro、JSON、Delta Lake、CSV
凭证管理	Azure Key Vault

🚀 路线图

• 更多 Feathr 在线客户端库，例如 Java
• 支持特征版本控制
• 支持特征监控

👨‍👨‍👦‍👦 社区准则

为社区而建，由社区共建。请查看社区准则。

📢 Slack 社区频道

加入我们的Slack 社区频道，进行问题交流和讨论（或点击邀请链接）。

Feathr 快速上手指南

Feathr 是一个可扩展、统一的企业级数据与 AI 工程平台（特征存储），源自 LinkedIn 生产环境，现由 LF AI & Data 基金会托管。它支持使用 Pythonic API 定义特征转换、注册特征，并自动处理离线批处理、流式计算及在线服务中的时间点正确性（Point-in-time Correctness），有效防止数据泄露。

1. 环境准备

在开始之前，请确保您的开发环境满足以下要求：

• 操作系统：Linux, macOS 或 Windows (推荐 WSL2)。
• Python 版本：Python 3.8 - 3.10。
• 前置依赖：
  • pip (Python 包管理工具)
  • Docker (如果使用 Sandbox 模式快速体验，必须安装)
  • Java (若本地运行 Spark 相关功能可能需要，Sandbox 中已内置)

2. 安装步骤

您可以根据需求选择以下两种方式进行安装：

方式一：使用 Feathr Sandbox（推荐新手快速体验）

这是最快捷的方式，通过 Docker 容器运行一个包含 Feathr UI、Jupyter Notebook 和后端服务的完整环境，无需配置任何云资源。

启动命令：

docker run -it --rm -p 8888:8888 -p 8081:80 -p 7080:7080 -e GRANT_SUDO=yes feathrfeaturestore/feathr-sandbox:releases-v1.0.0

启动成功后：

1. 编写代码：访问 http://localhost:8888/lab/workspaces/auto-w/tree/local_quickstart_notebook.ipynb 打开预置的 Jupyter Notebook。
2. 查看界面：访问 http://localhost:8081 查看 Feathr UI，用于浏览和搜索已注册的特征。

方式二：本地安装 Python 客户端

如果您希望在现有 Python 环境中集成 Feathr 客户端以连接云端（如 Azure Synapse, Databricks）或本地 Spark 集群：

使用 PyPI 安装（稳定版）：

pip install feathr

使用 GitHub 源码安装（最新版）：

pip install git+https://github.com/feathr-ai/feathr.git#subdirectory=feathr_project

提示：国内用户若下载缓慢，可尝试配置清华源或阿里源： pip install feathr -i https://pypi.tuna.tsinghua.edu.cn/simple

3. 基本使用

以下示例展示如何使用 Feathr 定义数据源、创建基于时间窗口的聚合特征以及派生特征。

步骤 1：定义数据源与预处理 (UDF)

Feathr 支持原生 PySpark 和 Spark SQL 进行数据预处理。

from pyspark.sql.functions import dayofweek
from feathr import HdfsSource, FeatureAnchor, Feature, WindowAggTransformation, DerivedFeature
from feathr.dtype import FLOAT, DENSE_VECTOR

# 定义预处理函数 (UDF)
def add_new_dropoff_and_fare_amount_column(df):
    df = df.withColumn("f_day_of_week", dayofweek("lpep_dropoff_datetime"))
    df = df.withColumn("fare_amount_cents", df.fare_amount.cast('double') * 100)
    return df

# 定义批量数据源
batch_source = HdfsSource(name="nycTaxiBatchSource",
                          path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/demo_data/green_tripdata_2020-04.csv",
                          preprocessing=add_new_dropoff_and_fare_amount_column,
                          event_timestamp_column="new_lpep_dropoff_datetime",
                          timestamp_format="yyyy-MM-dd HH:mm:ss")

步骤 2：定义窗口聚合特征

定义具有时间点正确性的窗口聚合特征（例如：过去 90 天的平均车费）。

# 假设 location_id 已在前文定义为 key
agg_features = [Feature(name="f_location_avg_fare",
                        key=location_id,
                        feature_type=FLOAT,
                        transform=WindowAggTransformation(
                            agg_expr="cast_float(fare_amount)",
                            agg_func="AVG",
                            window="90d"))]

agg_anchor = FeatureAnchor(name="aggregationFeatures",
                           source=batch_source,
                           features=agg_features)

步骤 3：定义派生特征

基于现有特征计算新特征（例如：结合距离和时间计算相似度或衍生值）。

# 假设 f_trip_distance 和 f_trip_time_duration 是已定义的特征
derived_feature = DerivedFeature(name="f_trip_time_distance",
                                 feature_type=FLOAT,
                                 key=trip_key,
                                 input_features=[f_trip_distance, f_trip_time_duration],
                                 transform="f_trip_distance * f_trip_time_duration")

后续步骤

定义好特征后，您可以在本地 Sandbox 或云端环境中执行以下操作：

1. 注册特征：将特征元数据注册到 Feathr Registry。
2. 材料化 (Materialize)：将特征计算结果写入离线存储或在线数据库。
3. 获取特征：在模型训练或在线推理时，通过 Point-in-time Join 获取准确的特征数据。

详细的生产环境部署（Azure/Databricks）请参考官方文档中的 Deployment Guide。