如何使用AWS胶水/火花将S3中的CSVs分区和分割转换为分区和分割拼花地板

Question

s3://my-data-lake/test-table/
    2017/01/01/
        part-0000-blah.csv.gz
        .
        .
        part-8000-blah.csv.gz
    2017/01/02/
        part-0000-blah.csv.gz
        .
        .
        part-7666-blah.csv.gz

我如何使用胶水/火花转换成拼花，这也是分区的日期和分裂在n个文件每天？。这些示例不包括分区、拆分或供应（多少节点和多大节点）。每天包含几百GBS。

因为源CSV不一定在正确的分区中（错误的日期），并且大小不一致，所以我希望用正确的分区和更一致的大小写到分区的parquet。

Answer 1

由于源CSV文件的日期不一定正确，您可以向它们添加关于收集日期时间的附加信息（或者使用任何已经可用的日期）：

{"collectDateTime": {
    "timestamp": 1518091828,
    "timestampMs": 1518091828116,
    "day": 8,
    "month": 2,
    "year": 2018
}}

然后，您的作业可以在输出DynamicFrame中使用这些信息，并最终将它们用作分区。如何实现这一点的一些示例代码：

from awsglue.transforms import *
from pyspark.sql.types import *
from awsglue.context import GlueContext
from awsglue.utils import getResolvedOptions

import sys
import datetime

###
# CREATE THE NEW SIMPLIFIED LINE
##
def create_simplified_line(event_dict):

    # collect date time
    collect_date_time_dict = event_dict["collectDateTime"]

    new_line = {
        # TODO: COPY YOUR DATA HERE
        "myData": event_dict["myData"],
        "someOtherData": event_dict["someOtherData"],
        "timestamp": collect_date_time_dict["timestamp"],
        "timestampmilliseconds": long(collect_date_time_dict["timestamp"]) * 1000,
        "year": collect_date_time_dict["year"],
        "month": collect_date_time_dict["month"],
        "day": collect_date_time_dict["day"]
    }

    return new_line


###
# MAIN FUNCTION
##

# context
glueContext = GlueContext(SparkContext.getOrCreate())

# fetch from previous day source bucket
previous_date = datetime.datetime.utcnow() - datetime.timedelta(days=1)

# build s3 paths
s3_path = "s3://source-bucket/path/year={}/month={}/day={}/".format(previous_date.year, previous_date.month, previous_date.day)

# create dynamic_frame
dynamic_frame = glueContext.create_dynamic_frame.from_options(connection_type="s3", connection_options={"paths": [s3_path]}, format="json", format_options={}, transformation_ctx="dynamic_frame")

# resolve choices (optional)
dynamic_frame_resolved = ResolveChoice.apply(frame=dynamic_frame,choice="project:double",transformation_ctx="dynamic_frame_resolved")

# transform the source dynamic frame into a simplified version
result_frame = Map.apply(frame=dynamic_frame_resolved, f=create_simplified_line)

# write to simple storage service in parquet format
glueContext.write_dynamic_frame.from_options(frame=result_frame, connection_type="s3", connection_options={"path":"s3://target-bucket/path/","partitionKeys":["year", "month", "day"]}, format="parquet")

没有测试它，但脚本只是如何实现这一点的一个样本，而且相当简单。

根据“partition_col”转换为dataframe和分区

partitioned_dataframe=datasource0.todf().repartitioned(1)

转换回DynamicFrame以进行进一步处理。

dynamic_frame = glueContext.create_dynamic_frame.from_options(connection_type="s3", connection_options={"paths": [s3_path], 'groupFiles': 'inPartition', 'groupSize': 1024 * 1024}, format="json", format_options={}, transformation_ctx="dynamic_frame")

值得一提的是，这与coalesce不同，但如果您的目标是减少每个分区的文件数量，它可能会有所帮助。

2）如果目标中已经存在文件，它是否会安全地添加它（而不是覆盖或删除）

Glue对write_dynamic_frame.from_options的默认SaveMode是追加。