apache-spark - 使用 Amazon Glue 将一行转换为多行

Question

我正在尝试使用 Amazon Glue 将一行变成多行。我的目标类似于 SQL UNPIVOT。

我有一个 360GB 的管道分隔文本文件，压缩 (gzip)。它有超过 1,620 列。这是基本布局:

primary_key|property1_name|property1_value|property800_name|property800_value
12345|is_male|1|is_college_educated|1

这些属性名称/值字段超过 800 个。大约有 2.8 亿行。该文件位于 S3 存储桶中。我需要将数据导入 Redshift，但 Redshift 中的列限制为 1,600。

用户希望我取消透视数据。例如:

primary_key|key|value
12345|is_male|1
12345|is_college_educated|1

我相信我可以为此使用 Amazon Glue。但是，这是我第一次使用 Glue。我正在努力寻找一个好的方法来做到这一点。一些 pySpark 扩展转换看起来很有希望(可能是“映射”或“关系化”)。见 http://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-etl-scripts-pyspark-transforms.html . 所以，我的问题是:在 Glue 中执行此操作的好方法是什么？

谢谢。

Answer 1

AWS Glue 没有适当的内置 GlueTransform 子类来将单个 DynamicRecord 转换为多个(通常 MapReduce 映射器可以这样做)。您也不能自己创建这样的转换。

但是有两种方法可以解决您的问题。

选项 1:使用 Spark RDD API

让我们尝试执行您需要的操作:将单个记录映射到多个记录。由于 GlueTransform 的限制，我们将不得不深入研究并使用 Spark RDD API。

RDD 有特殊的 flatMap允许生成多个 Row 的方法，然后将其展平。您的示例代码如下所示:

source_data = somehow_get_the_data_into_glue_dynamic_frame()
source_data_rdd = source_data.toDF().rdd
unpivoted_data_rdd = source_data_rdd.flatMap(
    lambda row: (
        (
            row.id,
            getattr(row, f'{field}_name'),
            getattr(row, f'{field}_value'),
        )
        for field in properties_names
    ),
)
unpivoted_data = glue_ctx.create_dynamic_frame \
    .from_rdd(unpivoted_data_rdd, name='unpivoted')

选项 2:映射 + 关系化 + 连接

如果您只想使用 AWS Glue ETL API 执行请求的操作，那么这里是我的说明:

1. 第一 map从源到主键和对象列表的每一个 DynamicRecord:

mapped = Map.apply(
    source_data,
    lambda record:  # here we operate on DynamicRecords not RDD Rows
        DynamicRecord(
            primary_key=record.primary_key,
            fields=[
                dict(
                    key=getattr(row, f'{field}_name'),
                    value=getattr(row, f'{field}_value'),
                )
                for field in properties_names
            ],
        )
)

示例输入:

primary_key|property1_name|property1_value|property800_name|property800_value
      12345|is_male       |              1|is_new          |                1
      67890|is_male       |              0|is_new          |                0

输出:

primary_key|fields
      12345|[{'key': 'is_male', 'value': 1}, {'key': 'is_new', 'value': 1}]
      67890|[{'key': 'is_male', 'value': 0}, {'key': 'is_new', 'value': 0}]

1. 下一个 relationalize它:每个列表都将转换为多行，每个嵌套对象都将被取消嵌套(Scala Glue ETL API docs 有很好的示例和比 Python 文档更详细的解释)。

relationalized_dfc = Relationalize.apply(
    mapped,
    staging_path='s3://tmp-bucket/tmp-dir/',  # choose any dir for temp files
)

该方法返回DynamicFrameCollection。在单个数组字段的情况下，它将包含两个 DynamicFrame:首先是 primary_key 和扁平化和未嵌套 fields 动态帧的外键。 输出:

# table name: roottable
primary_key|fields
      12345|     1
      67890|     2

# table name: roottable.fields
id|index|val.key|val.value
 1|    0|is_male|        1
 1|    1|is_new |        1
 2|    0|is_male|        0
 2|    1|is_new |        0

1. 最后一个合乎逻辑的步骤是加入这两个 DynamicFrame 的:

joined = Join.apply(
    frame1=relationalized_dfc['roottable'],
    keys1=['fields'],
    frame2=relationalized_dfc['roottable.fields'],
    keys2=['id'],
)

输出:

primary_key|fields|id|index|val.key|val.value
      12345|     1| 1|    0|is_male|        1
      12345|     1| 1|    1|is_new |        1
      67890|     2| 2|    0|is_male|        0
      67890|     2| 2|    1|is_new |        0

现在您只需 rename和 select所需的字段。