AWS Glue 书签答案

【问题标题】：AWS Glue BookmarksAWS Glue 书签
【发布时间】：2019-05-12 14:23:42
【问题描述】：

如何验证我的书签是否正常工作？我发现当我在之前完成后立即运行一项工作时，似乎仍然需要很长时间。这是为什么？我以为它不会读取它已经处理的文件？脚本如下所示：

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

inputGDF = glueContext.create_dynamic_frame_from_options(connection_type = "s3", connection_options = {"paths": ["s3://xxx-glue/testing-csv"], "recurse": True}, format = "csv", format_options = {"withHeader": True}, transformation_ctx="inputGDF")

if bool(inputGDF.toDF().head(1)):
    print("Writing ...")
    inputGDF.toDF() \
      .drop("createdat") \
      .drop("updatedat") \
      .write \
      .mode("append") \
      .partitionBy(["querydestinationplace", "querydatetime"]) \
      .parquet("s3://xxx-glue/testing-parquet")
else:
    print("Nothing to write ...")

job.commit()

import boto3
glue_client = boto3.client('glue', region_name='ap-southeast-1')
glue_client.start_crawler(Name='xxx-testing-partitioned')

日志如下：

18/12/11 14:49:03 INFO Client: Application report for application_1544537674695_0001 (state: RUNNING)
18/12/11 14:49:03 DEBUG Client: 
client token: N/A
diagnostics: N/A
ApplicationMaster host: 172.31.2.72
ApplicationMaster RPC port: 0
queue: default
start time: 1544539297014
final status: UNDEFINED
tracking URL: http://ip-172-31-0-204.ap-southeast-1.compute.internal:20888/proxy/application_1544537674695_0001/
user: root
18/12/11 14:49:04 INFO Client: Application report for application_1544537674695_0001 (state: RUNNING)
18/12/11 14:49:04 DEBUG Client: 
client token: N/A
diagnostics: N/A
ApplicationMaster host: 172.31.2.72
ApplicationMaster RPC port: 0
queue: default
start time: 1544539297014
final status: UNDEFINED
tracking URL: http://ip-172-31-0-204.ap-southeast-1.compute.internal:20888/proxy/application_1544537674695_0001/
user: root
18/12/11 14:49:05 INFO Client: Application report for application_1544537674695_0001 (state: RUNNING)
18/12/11 14:49:05 DEBUG Client: 
client token: N/A
diagnostics: N/A
ApplicationMaster host: 172.31.2.72
ApplicationMaster RPC port: 0
queue: default
start time: 1544539297014
final status: UNDEFINED
tracking URL: http://ip-172-31-0-204.ap-southeast-1.compute.internal:20888/proxy/application_1544537674695_0001/
user: root
...

18/12/11 14:42:00 INFO NewHadoopRDD: Input split: s3://pinfare-glue/testing-csv/2018-09-25/DPS/2018-11-15_2018-11-19.csv:0+1194081
18/12/11 14:42:00 INFO S3NativeFileSystem: Opening 's3://pinfare-glue/testing-csv/2018-09-25/DPS/2018-11-14_2018-11-18.csv' for reading
18/12/11 14:42:00 INFO S3NativeFileSystem: Opening 's3://pinfare-glue/testing-csv/2018-09-25/DPS/2018-11-15_2018-11-19.csv' for reading
18/12/11 14:42:00 INFO Executor: Finished task 89.0 in stage 0.0 (TID 89). 2088 bytes result sent to driver
18/12/11 14:42:00 INFO CoarseGrainedExecutorBackend: Got assigned task 92
18/12/11 14:42:00 INFO Executor: Running task 92.0 in stage 0.0 (TID 92)
18/12/11 14:42:00 INFO NewHadoopRDD: Input split: s3://pinfare-glue/testing-csv/2018-09-25/DPS/2018-11-16_2018-11-20.csv:0+1137753
18/12/11 14:42:00 INFO Executor: Finished task 88.0 in stage 0.0 (TID 88). 2088 bytes result sent to driver
18/12/11 14:42:00 INFO CoarseGrainedExecutorBackend: Got assigned task 93
18/12/11 14:42:00 INFO Executor: Running task 93.0 in stage 0.0 (TID 93)
18/12/11 14:42:00 INFO NewHadoopRDD: Input split: s3://pinfare-glue/testing-csv/2018-09-25/DPS/2018-11-17_2018-11-21.csv:0+1346626
18/12/11 14:42:00 INFO S3NativeFileSystem: Opening 's3://pinfare-glue/testing-csv/2018-09-25/DPS/2018-11-16_2018-11-20.csv' for reading
18/12/11 14:42:00 INFO S3NativeFileSystem: Opening 's3://pinfare-glue/testing-csv/2018-09-25/DPS/2018-11-17_2018-11-21.csv' for reading
18/12/11 14:42:00 INFO Executor: Finished task 90.0 in stage 0.0 (TID 90). 2088 bytes result sent to driver
18/12/11 14:42:00 INFO Executor: Finished task 91.0 in stage 0.0 (TID 91). 2088 bytes result sent to driver
18/12/11 14:42:00 INFO CoarseGrainedExecutorBackend: Got assigned task 94
18/12/11 14:42:00 INFO CoarseGrainedExecutorBackend: Got assigned task 95
18/12/11 14:42:00 INFO Executor: Running task 95.0 in stage 0.0 (TID 95)
18/12/11 14:42:00 INFO Executor: Running task 94.0 in stage 0.0 (TID 94)

...我注意到镶木地板上附加了很多重复数据...书签不起作用吗？它已经启用

【问题讨论】：

标签： amazon-web-services pyspark parquet aws-glue

【解决方案1】：

书签要求

来自the docs

必须使用 --job-bookmark-option job-bookmark-enable 创建作业（或者如果使用控制台，则在控制台选项中）。工作也必须有工作名称；这会自动传入。
作业必须以job.init(jobname) 开头例如

job = Job(glueContext)
job.init(args['JOB_NAME'], args)

作业必须有一个job.commit() 才能保存书签的状态并成功完成。
数据源必须是 s3 源或 JDBC（有限，而不是您的用例，所以我将忽略它）。

example in the docs 显示使用表名而不是显式 S3 路径从（Glue/Lake 形成）目录创建动态框架。这意味着从目录中读取仍被视为 S3 源；底层文件将在 S3 上。

对于 0.9 及以上版本，s3 上的文件必须是 JSON、CSV、Apache Avro、XML 之一，对于 1.0 及以上版本，可以是 Parquet 或 ORC
脚本中的数据源必须有transformation_ctx参数。
文档说

仅将 transformation_ctx 参数传递给您想要启用书签您可以将其添加到每个转换中以保存状态，但关键的是您要添加书签的数据源。

疑难解答

来自the docs

最大并发数必须为 1。更高的值会破坏书签
上面还提到了job.commit() 并使用transformation_ctx

对于 Amazon S3 输入源，作业书签检查上次修改 对象的时间，而不是文件名，以验证哪个对象需要重新处理。如果您的输入源数据已自上次作业运行以来已修改，当您重新处理文件时再次运行作业。

其他需要检查的事项

您是否已验证路径 "s3://xxx-glue/testing-csv" 中的 CSV 文件不包含重复项？您可以使用 Glue 爬虫或在 Athena 中编写 DDL 在它们之上创建一个表并直接查看。或者，创建一个开发端点并运行一个 zeppelin 或 sagemaker 笔记本并单步执行您的代码。
它没有提到编辑脚本会重置您的状态的任何地方，但是，如果您修改了数据源或其他阶段的transformation_ctx，那么这可能会影响状态，但是我还没有验证这一点。该作业有一个Jobname，它是状态的关键，以及用于管理重试和最新状态的运行号、尝试号和版本号，这意味着对脚本的微小更改不会影响状态只要因为Jobname 是一致的，但我还没有验证。
顺便说一句，在您的代码中测试inputGDF.toDF().head(1)，然后运行inputGDF.toDF()... 来写入数据。 Spark 被延迟评估，但在这种情况下，您将两次运行与数据帧等效的动态帧，并且 spark 无法缓存或重用它。最好在if 之前执行df = inputGDF.toDF() 之类的操作，然后重复使用df 两次。

【讨论】：

嗨@Davos，我不想使用spark。我的文件在 S3 中，所以我可以使用在 Spark Job 上运行的简单 python 来做到这一点吗？
@Abdul_Haseeb 这个问题有很多潜在的答案，这需要更多的背景知识，也许是一个全新的问题。如果你想在 Spark 集群上使用 Python，你可以，但你会失去 Spark 提供的分布式处理的好处。
是的，我知道，我将失去分布式处理，现在我正在使用 SQS 跟踪 s3 上新上传的文件，但我想粘贴书签功能来跟踪。
您需要使用transformation_ctx 和job.commit() 部分来执行数据源，但您可以在中间执行任何您喜欢的操作。数据源将为您提供一个火花数据帧（或胶水“动态帧”包装器），因此如果您想使用 python，您可以从那里将其转换为熊猫（这是一个缓慢的操作，因为所有文件都必须收集到驱动程序请参阅docs.databricks.com/spark/latest/spark-sql/spark-pandas.html) 或者您可以将 python 函数注册为 UDF，然后在 Pyspark 中使用它们。仅仅为了获得书签功能就付出了很多努力。
您多次错过拼写 job.commit()。没有大写字母和 2 个“m”。我愚蠢地复制/粘贴了您的一些代码，但失败了，我花了几分钟才弄清楚原因。

【解决方案2】：

Glue version 0.9 中的 parquet 格式不支持书签：

Glue version 1.0 支持它们。

【讨论】：

0.9 版确实如此，但文档已更新。从 1.0 版开始支持 Parquet 和 ORC

【解决方案3】：

https://docs.aws.amazon.com/glue/latest/dg/monitor-debug-multiple.html 可用于验证书签是否工作

【讨论】：

【解决方案4】：

请查看 this 文档了解 AWS Glue 书签机制。

基本上它需要通过控制台（或 CloudFormation）启用它并指定tranformation_context 参数，该参数与其他一些属性（如作业名称、源文件名）一起使用以保存检查点信息。如果您更改这些属性之一的值，则 Glue 会将其视为不同的检查点。

【讨论】：

【解决方案5】：

只是为了记录，因为还没有答案。

我认为编辑脚本似乎会影响书签...但我认为它不应该...

【讨论】：