AWS Glue Crawl 动态 S3 路径位置

Question

我正在 AWS Glue 中创建一个 ETL 作业，该作业将从 S3 位置提取存储库中每个实体的最新编辑或当前数据。存储库中的数据是实体所有编辑的历史帐户。每天我运行 ETL 并写入另一个 S3 位置，即存储桶/路径/到/文件/current_date/...其中当前日期是动态的，并且与 ETL 运行的日期一致。

我遇到的问题是我无法从 S3(组织限制)中以编程方式删除文件，或者移动文件，因为这是副本并在后台删除，因此它也失败了，留下一条路径供胶水爬行。 我想设置爬虫，使路径的日期部分是动态的，但我无法找到一种方法来做到这一点 - 有谁知道这是否可能？

我的数据按run_date(请参阅上面的当前日期)以及其他 6 个分层分区进行分区。我正在通过CloudFormation，yaml语言创建爬虫和ETL作业。 爬网程序的路径存储为 CloudFormation 脚本中定义的 ssm 参数。

路径 SSM 参数示例

S3CurrentPath:
Type: AWS::SSM::Parameter
Properties:
Description: "Path in the S3 Lake where the current entity data is stored."
Type: String
Value: 'Data/Entities/Software/SoftwareCurrent'
Name: "/org/member/local/s3/path/entityCurrent"

爬网程序资源代码：

GenericCrawler:
Type: AWS::Glue::Crawler
Properties:
Role: !Ref RoleNAme
Name: !Sub "${ProfileName}-crawler-${CrawlerName}"
Configuration: !Sub |
{
"Version": 1.0,
"CrawlerOutput": {
"Partitions": { "AddOrUpdateBehavior": "InheritFromTable" },
"Tables": { "AddOrUpdateBehavior": "MergeNewColumns" }
}
}
Description: !Ref CrawlerDescription
DatabaseName: !Ref DatabaseName
Targets:
S3Targets:
- Path: !Sub "s3://${S3DataBucket}/${S3Path}"

ETL 数据接收器写入代码：

# Write the joined dynamic frame out to a datasink
datasink = glueContext.write_dynamic_frame.from_options(
frame = final_dynamic_frame, connection_type = "s3",
connection_options = {
'path': 's3://{lakeBucketName}/{lakePath}/'.format(
lakeBucketName=args['lakeBucketName'],
lakePath=args['lakeDestinationPath']),
"partitionKeys": ['run_date','location','year','month','day','hour','timestamp']},
format = "parquet",
transformation_ctx = "datasink")

我希望爬虫会查看存储库中的最新日期，即最近的run_date分区"文件夹"并对其进行爬网，而无需查看旧数据。

如果您想查看更多代码，请告诉我 - 我很乐意进行消毒和提供。

Answer 1

老实说，我还没有找到使用 AWS Glue 将数据读取/写入动态路径的方法。我通常做的是使用 PySpark 方法读/写：

datasink.write.
format("com.databricks.spark.csv").
option("header", "true").
mode("overwrite").
save("s3://my-bucket/files/" + current_date + "*.csv")

您甚至可以告诉该方法只是读取/写入特定类型的文件(例如.csv)。 与 AWS Glue 相比，PySpark 具有更多的选项和可用方法，因此具有更大的灵活性。此外，我还在 DynamoDB 表中添加了键/值记录，以保留最新日期的记录。