Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

使用亚马逊 Feature Store 批量摄取 Sp SageMaker ark

Amazon F SageMaker eature Store Spark 是一个 Spark 连接器，它将 Spark 库连接到功能商店。Feature Store Spark 简化了从 Spark DataFrame 到特征组的数据摄取流程。Feature Store 支持使用现有的 ETL 管道，在 Amazon EMR、GIS、作业 Amazon Glue 、亚马逊 SageMaker 处理任务或笔记本上使用 Spark 进行批量数据摄取。 SageMaker

我们为 Python 和 Scala 开发人员提供了安装和实施批量数据摄取的方法。Python 开发者可以按照亚马逊 SageMaker 功能商店 Spark 存储库中的说明使用开源 sagemaker-feature-store-pyspark Python 库进行本地开发、在 Amazon EMR 上安装以及 Jupyter Notebook。 GitHubScala 开发人员可以使用亚马逊功能商店 Spark SDK Maven 中央存储库中提供的 SageMaker 功能商店 Spark 连接器。

可以通过以下方式使用 Spark 连接器摄取数据，具体取决于是启用了在线存储、离线存储，还是两者均已启用。

有关使用不同摄取方法的信息，请参阅示例实施。

Feature Store Spark 安装

Scala 用户

Feature Store Spark SDK 可在亚马逊 SageMaker 功能商店 Spark SDK Maven 中央存储库中获得，供 Scala 用户使用。

要求

在 POM.xml 中声明依赖项

Feature Store Spark 连接器在 iceberg-spark-runtime 库中有一个依赖项。因此，如果要将数据摄取到使用 Iceberg 表格式自动创建的特征组中，则必须将相应版本的 iceberg-spark-runtime 库添加到该依赖项中。例如，如果使用的是 Spark 3.1，则必须在项目的 POM.xml 中声明以下内容：

 <dependency>
 <groupId>software.amazon.sagemaker.featurestore</groupId>
 <artifactId>sagemaker-feature-store-spark-sdk_2.12</artifactId>
 <version>1.0.0</version>
 </dependency>
 
 <dependency>
   <groupId>org.apache.iceberg</groupId>
   <artifactId>iceberg-spark-runtime-3.1_2.12</artifactId>
   <version>0.14.0</version>
</dependency>
    

Python 用户

Feature Store Spark SDK 可在开源亚马逊 SageMaker 功能商店 Spark GitHub 存储库中找到。

要求

我们建议将 $SPARK_HOME 设置为安装了 Spark 的目录。安装期间，Feature Store 会将所需的 JAR 上传到 SPARK_HOME，这样就会自动加载依赖项。要使该 PySpark 库正常工作，需要启动 Spark 启动 JVM。

本地安装

要查找有关安装的更多信息，请通过将 --verbose 附加到以下安装命令来启用详细模式。

pip3 install sagemaker-feature-store-pyspark-3.1 --no-binary :all:
    

在 Amazon EMR 上安装

使用发行版 6.1.0 或更高版本创建一个 Amazon EMR 集群。启用 SSH 可帮助您解决任何问题。

可以执行以下操作之一来安装库：

export SPARK_HOME=/usr/lib/spark
sudo -E pip3 install sagemaker-feature-store-pyspark-3.1 --no-binary :all: --verbose      
    

在 SageMaker 笔记本实例上安装

使用以下命令安装 PySpark 与 Spark 连接器兼容的版本：

!pip3 install pyspark==3.1.1 
!pip3 install sagemaker-feature-store-pyspark-3.1 --no-binary :all:  

如果您要对离线存储执行批量摄取，则依赖项不在笔记本实例环境中。

from pyspark.sql import SparkSession
import feature_store_pyspark

extra_jars = ",".join(feature_store_pyspark.classpath_jars())

spark = SparkSession.builder \
    .config("spark.jars", extra_jars) \
    .config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.2.1,org.apache.hadoop:hadoop-common:3.2.1") \
    .getOrCreate()
    

在带有 GIS 的笔记本上安装

使用以下信息帮助您在 Amazon Glue 交互式会话 (GIS) 中安装 PySpark 连接器。

Amaz SageMaker on Feature Store Spark 需要在会话初始化期间使用特定的 Spark 连接器 JAR 上传到您的 Amazon S3 存储桶。有关将所需 JAR 上传到 S3 存储桶的更多信息，请参阅检索 Feature Store Spark 的 JAR。

上传 JAR 后，必须使用以下命令为 GIS 会话提供 JAR。

%extra_jars s3:/<YOUR_BUCKET>/spark-connector-jars/sagemaker-feature-store-spark-sdk.jar

要在 Amazon Glue 运行时安装 Feature Store Spark，请使用 GIS 笔记本中的%additional_python_modules神奇命令。 Amazon Glue 运行pip到您在下指定的模块%additional_python_modules。

%additional_python_modules sagemaker-feature-store-pyspark-3.1

在开始会 Amazon Glue 话之前，必须使用前面的两个神奇命令。

在 Amazon Glue 工作中安装

要在 Amazon Glue 作业上安装 Spark 连接器，请使用--extra-jars参数提供必要的 JAR，并在--additional-python-modules创建作业时将 Spark 连接器作为 Amazon Glue 作业参数安装，如下例所示。有关将所需 JAR 上传到 S3 存储桶的更多信息，请参阅检索 Feature Store Spark 的 JAR。

glue_client = boto3.client('glue', region_name=region)
response = glue_client.create_job(
    Name=pipeline_id,
    Description='Feature Store Compute Job',
    Role=glue_role_arn,
    ExecutionProperty={'MaxConcurrentRuns': max_concurrent_run},
    Command={
        'Name': 'glueetl',
        'ScriptLocation': script_location_uri,
        'PythonVersion': '3'
    },
    DefaultArguments={
        '--TempDir': temp_dir_location_uri,
        '--additional-python-modules': 'sagemaker-feature-store-pyspark-3.1',
        '--extra-jars': "s3:/<YOUR_BUCKET>/spark-connector-jars/sagemaker-feature-store-spark-sdk.jar",
        ...
    },
    MaxRetries=3,
    NumberOfWorkers=149,
    Timeout=2880,
    GlueVersion='3.0',
    WorkerType='G.2X'
)
    

在 Amazon SageMaker 处理任务上安装

要将 Feature Store Spark 与 Amazon SageMaker 处理任务一起使用，请自带图片。有关自带映像的更多信息，请参阅带上你自己的 SageMaker 图片。向 Dockerfile 添加安装步骤。将 Docker 映像推送到 Amazon ECR 存储库后，您可以使用 PySparkProcessor 来创建处理任务。有关使用处理 PySpark 器创建处理任务的更多信息，请参阅使用 Apache Spark 进行数据处理。

以下是向 Dockerfile 添加安装步骤的示例。

FROM <ACCOUNT_ID>.dkr.ecr.<AWS_REGION>.amazonaws.com/sagemaker-spark-processing:3.1-cpu-py38-v1.0

RUN /usr/bin/python3 -m pip install sagemaker-feature-store-pyspark-3.1 --no-binary :all: --verbose
      
    

检索 Feature Store Spark 的 JAR

要检索 Feature Store Spark 依赖项 JAR，必须在任何具有网络访问权限的 Python 环境中使用 pip 从 Python Package Index (PyPI) 存储库中安装 Spark 连接器。 SageMaker Jupyter 笔记本就是一个具有网络访问权限的 Python 环境的示例。

以下命令用于安装 Spark 连接器。

!pip install sagemaker-feature-store-pyspark-3.1      
    

安装 Feature Store Spark 后，您可以检索 JAR 位置并将 JAR 上传到 Amazon S3。

feature-store-pyspark-dependency-jars 命令提供了 Feature Store Spark 添加的必要依赖项 JAR 的位置。您可以使用该命令来检索 JAR，并将其上传到 Amazon S3。

jar_location = !feature-store-pyspark-dependency-jars
jar_location = jar_location[0]

s3_client = boto3.client("s3")
s3_client.upload_file(jar_location, "<YOUR_BUCKET>","spark-connector-jars/sagemaker-feature-store-spark-sdk.jar")
      

示例实施

Example Python script

FeatureStoreBatchIngestion.py

from pyspark.sql import SparkSession
from feature_store_pyspark.FeatureStoreManager import FeatureStoreManager
import feature_store_pyspark

spark = SparkSession.builder \
                    .getOrCreate()

# Construct test DataFrame
columns = ["RecordIdentifier", "EventTime"]
data = [("1","2021-03-02T12:20:12Z"), ("2", "2021-03-02T12:20:13Z"), ("3", "2021-03-02T12:20:14Z")]

df = spark.createDataFrame(data).toDF(*columns)

# Initialize FeatureStoreManager with a role arn if your feature group is created by another account
feature_store_manager= FeatureStoreManager("arn:aws:iam::111122223333:role/role-arn")
 
# Load the feature definitions from input schema. The feature definitions can be used to create a feature group
feature_definitions = feature_store_manager.load_feature_definitions_from_schema(df)

feature_group_arn = "arn:aws:sagemaker:<AWS_REGION>:<ACCOUNT_ID>:feature-group/<YOUR_FEATURE_GROUP_NAME>"

# Ingest by default. The connector will leverage PutRecord API to ingest your data in stream
# https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_feature_store_PutRecord.html
feature_store_manager.ingest_data(input_data_frame=df, feature_group_arn=feature_group_arn)

# To select the target stores for ingestion, you can specify the target store as the paramter
# If OnlineStore is selected, the connector will leverage PutRecord API to ingest your data in stream
feature_store_manager.ingest_data(input_data_frame=df, feature_group_arn=feature_group_arn, target_stores=["OfflineStore", "OnlineStore"])

# If only OfflineStore is selected, the connector will batch write the data to offline store directly
feature_store_manager.ingest_data(input_data_frame=df, feature_group_arn=feature_group_arn, target_stores=["OfflineStore"])

# To retrieve the records failed to be ingested by spark connector
failed_records_df = feature_store_manager.get_failed_stream_ingestion_data_frame()
    

使用示例 Python 脚本提交 Spark 作业

该 PySpark 版本需要导入额外的依赖 JAR，因此需要额外的步骤来运行 Spark 应用程序。

如果您在安装期间未指定 SPARK_HOME，则在运行 spark-submit 时必须在 JVM 中加载所需的 JAR。feature-store-pyspark-dependency-jars 是 Spark 库安装的 Python 脚本，用于自动为您获取所有 JAR 的路径。

spark-submit --jars `feature-store-pyspark-dependency-jars` FeatureStoreBatchIngestion.py
    

如果要在 Amazon EMR 上运行此应用程序，我们建议您在客户端模式下运行该应用程序，这样就无需将从属 JAR 分配给其他任务节点。使用类似于以下内容的 Spark 参数在 Amazon EMR 集群中再添加一个步骤：

spark-submit --deploy-mode client --master yarn s3:/<PATH_TO_SCRIPT>/FeatureStoreBatchIngestion.py
    

Example Scala script

FeatureStoreBatchIngestion.scala

import software.amazon.sagemaker.featurestore.sparksdk.FeatureStoreManager
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SparkSession}

object TestSparkApp {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder().getOrCreate()

    // Construct test DataFrame
    val data = List(
      Row("1", "2021-07-01T12:20:12Z"),
      Row("2", "2021-07-02T12:20:13Z"),
      Row("3", "2021-07-03T12:20:14Z")
    )
    
    val schema = StructType(
      List(StructField("RecordIdentifier", StringType), StructField("EventTime", StringType))
    )

    val df = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
    
    // Initialize FeatureStoreManager with a role arn if your feature group is created by another account
    val featureStoreManager = new FeatureStoreManager("arn:aws:iam::111122223333:role/role-arn")
    
    // Load the feature definitions from input schema. The feature definitions can be used to create a feature group
    val featureDefinitions = featureStoreManager.loadFeatureDefinitionsFromSchema(df)

    val featureGroupArn = "arn:aws:sagemaker:<AWS_REGION>:<ACCOUNT_ID>:feature-group/<YOUR_FEATURE_GROUP_NAME>"
   
    // Ingest by default. The connector will leverage PutRecord API to ingest your data in stream
    // https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_feature_store_PutRecord.html
    featureStoreManager.ingestData(df, featureGroupArn)
    
    // To select the target stores for ingestion, you can specify the target store as the paramter
    // If OnlineStore is selected, the connector will leverage PutRecord API to ingest your data in stream
    featureStoreManager.ingestData(df, featureGroupArn, List("OfflineStore", "OnlineStore"))
    
    // If only OfflineStore is selected, the connector will batch write the data to offline store directly
    featureStoreManager.ingestData(df, featureGroupArn, ["OfflineStore"])
    
    // To retrieve the records failed to be ingested by spark connector
    val failedRecordsDf = featureStoreManager.getFailedStreamIngestionDataFrame()
  }
}
    

提交 Spark 作业

Scala

您应该能够将 Feature Store Spark 作为一个正常的依赖项来使用。无需额外的指令即可在所有平台上运行该应用程序。