AWS Glue PySpark 转换参考

AWS Glue 提供了以下可在 PySpark ETL 操作中使用的内置转换。您的数据在一个称为 DynamicFrame 的数据结构中从转换传递到转换，该数据结构是 Apache Spark SQL DataFrame 的扩展。DynamicFrame 包含您的数据，并引用其架构来处理您的数据。

此外，其中的大多数转换也将作为 DynamicFrame 类的方法存在。更多相关信息，请参阅 DynamicFrame 转换。

• GlueTransform 基类

• ApplyMapping 类

• DropFields 类

• DropNullFields 类

• ErrorsAsDynamicFrame 类

• EvaluateDataQuality 类

• FillMissingValues 类

• Filter 类

• FindIncrementalMatches 类

• FindMatches 类

• FlatMap 类

• Join 类

• Map 类

• MapToCollection 类

• mergeDynamicFrame

• Relationalize 类

• RenameField 类

• ResolveChoice 类

• SelectFields 类

• SelectFromCollection 类

• Simplify_ddb_json 类

• Spigot 类

• SplitFields 类

• SplitRows 类

• Unbox 类

• UnnestFrame 类

数据集成转换

对于 AWS Glue 4.0 及更高版本，使用 key: --enable-glue-di-transforms, value: true 创建或更新任务参数。

示例任务脚本：

from pyspark.context import SparkContext
        
from awsgluedi.transforms import *
sc = SparkContext()

input_df = spark.createDataFrame(
    [(5,), (0,), (-1,), (2,), (None,)],
    ["source_column"],
)

try:
    df_output = math_functions.IsEven.apply(
        data_frame=input_df,
        spark_context=sc,
        source_column="source_column",
        target_column="target_column",
        value=None,
        true_string="Even",
        false_string="Not even",
    )
    df_output.show()   
except:
    print("Unexpected Error happened ")
    raise

使用笔记本的示例会话

%idle_timeout 2880
%glue_version 4.0
%worker_type G.1X
%number_of_workers 5
%region eu-west-1

%%configure
{
    "--enable-glue-di-transforms": "true"
}

from pyspark.context import SparkContext
from awsgluedi.transforms import *

sc = SparkContext()

input_df = spark.createDataFrame(
    [(5,), (0,), (-1,), (2,), (None,)],
    ["source_column"],
)

try:
    df_output = math_functions.IsEven.apply(
        data_frame=input_df,
        spark_context=sc,
        source_column="source_column",
        target_column="target_column",
        value=None,
        true_string="Even",
        false_string="Not even",
    )
    df_output.show()    
except:
    print("Unexpected Error happened ")
    raise

使用 AWS CLI 的示例会话

aws glue create-session --default-arguments "--enable-glue-di-transforms=true"

DI 转换：

• FlagDuplicatesInColumn 类

• FormatPhoneNumber 类

• FormatCase 类

• FillWithMode 类

• FlagDuplicateRows 类

• RemoveDuplicates 类

• MonthName 类

• IsEven 类

• CryptographicHash 类

• 解密类

• 加密类

• IntToIp 类

• IpToInt 类

Maven：将插件与 Spark 应用程序捆绑在一起

在本地开发 Spark 应用程序时，您可以通过在 Maven pom.xml 中添加插件依赖关系，将转换依赖项与 Spark 应用程序和 Spark 发行版（版本 3.3）捆绑在一起。

<repositories>
   ...
    <repository>
        <id>aws-glue-etl-artifacts</id>
        <url>https://aws-glue-etl-artifacts.s3.amazonaws.com/release/ </url>
    </repository>
</repositories>
...
<dependency>
    <groupId>com.amazonaws</groupId>
    <artifactId>AWSGlueTransforms</artifactId>
    <version>4.0.0</version>
</dependency>

您也可以直接从 AWS Glue Maven 构件下载二进制文件，并将它们包含在您的 Spark 应用程序中，如下所示。

#!/bin/bash
sudo wget -v https://aws-glue-etl-artifacts.s3.amazonaws.com/release/com/amazonaws/AWSGlueTransforms/4.0.0/AWSGlueTransforms-4.0.0.jar -P /usr/lib/spark/jars/