AWS Glue 抓取动态 S3 路径位置

用户1983682

我正在 AWS Glue 中创建一个 ETL 作业,该作业将从 S3 位置提取存储库中每个实体的最新编辑或当前数据。存储库中的数据是实体所有编辑的历史记录。我每天运行 ETL 并将其写出到另一个 S3 位置,即 Bucket/path/to/files/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 DataSink 写入代码:

# 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 分区“文件夹”,并在不回头查看旧数据的情况下对其进行爬网。

如果您想查看更多代码,请告诉我——我很乐意清理并提供。

艾达·马丁内斯

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

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

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

本文收集自互联网,转载请注明来源。

如有侵权,请联系 [email protected] 删除。

编辑于
0

我来说两句

0 条评论
登录 后参与评论

相关文章

AWS Glue-如何从S3抓取Kinesis Firehose输出文件夹

使用Glue从AWS RDS到S3的管道

AWS Glue-是否使用抓取工具

使用 AWS Glue 将 AWS Redshift 转换为 S3 Parquet 文件

将动态参数传递给AWS Glue

阵列中的AWS Glue动态框架列

从 S3 解压缩文件并将其写回 S3 的 AWS Glue 作业

适用于Mongo的AWS Glue S3中的Parquet文件

AWS Glue 數據從 S3 轉移到 Redshift

AWS Glue 增量爬取 S3 上不断到达的数据

存储在S3中的AWS Glue Crawlers和大型表

使用 AWS Glue 從 S3 讀取動態 DataTpes

在AWS Glue作业中写入S3时是否可以指定SSE:KMS密钥ID?

AWS Glue S3目标-创建20个文件的分区

如何通过S3 Events或AWS Lambda触发Glue ETL Pyspark作业?

AWS Glue不会删除或弃用现在已删除的S3数据生成的表

使用Scala中的AWS Glue在S3中加载CSV文件

AWS Glue作业以实木复合地板格式写入s3错误,未找到

在AWS Glue ETL作业中从S3加载分区的json文件

何时使用 Amazon Redshift 频谱通过 AWS Glue ETL 查询 Amazon S3 数据

处理嵌套 s3 文件夹中的分区 Parquet 文件的 AWS GLUE 作业失败

如何从Scala的Glue Job中的S3文件创建动态数据框?

在相同位置重命名 S3 对象以移除 Glue 书签

获取AWS Glue写入另一个AWS账户中的S3存储桶

aws_boto3:列出 s3 路径的内容

AWS Glue / Athena-S3-表分区

将文件上传到S3存储桶后,基于事件的AWS Glue Crawler触发器?

AWS Athena从GLUE爬网程序从S3输入csv创建的表中返回零记录

AWS Glue:如何将S3存储桶分区到多个红移表中