AWS Glue ETL 作业使用 Join.apply Vs SQL JOIN Query 获取最终数据帧

Question

我对 AWS 还很陌生，目前正在探索它。我希望就实施这项工作的最佳方式获得见解或建议。

我想从多个 mysql 表中获取数据。

• user_transaction
• user_loans
• promo_offers

为了获得决赛桌，我找到了以下两种方式。

方法一：

1. 然后为每个表创建目录，

user_transaction = glueContext.create_dynamic_frame.from_catalog(
           database="Org_Data_Lake",
           table_name="user_transaction"
       transformation_ctx = "user_transaction", 
       additional_options = {"jobBookmarkKeys":["transaction_id"],"jobBookmarksKeysSortOrder":"asc"})

user_loans = glueContext.create_dynamic_frame.from_catalog(
           database="Org_Data_Lake",
           table_name="user_loans")
promo_offers = glueContext.create_dynamic_frame.from_catalog(
           database="Org_Data_Lake",
           table_name="promo_offers")

1. 然后应用 Join.apply 获得最终结果

final_history = Join.apply(user_transaction,
                       Join.apply(user_loans, promo_offers, 'offer_id', 'offer_id'),
                       'user_loan_id', 'user_loan_id').drop_fields([.......])

1. 最后将所有数据放入S3

glueContext.write_dynamic_frame.from_options(frame = final_history,
          connection_type = "s3",
          connection_options = {"path": "s3://glue-sample-target/output-dir/final_history"},
          format = "parquet")

方法二：

1. 立即准备好 final_history 数据，

query = "t1.transaction_id, t1.status, t2.loan_status, t3.offer_amount 
     FROM user_transaction AS t1 
     JOIN user_loans AS t2 ON (t2.user_loan_id = t1.user_loan_id) 
     JOIN promo_offers AS t3 ON (t3.offer_id = t2.offer_id) 
     WHERE t1.created_at > '2020-01-01 00:00:00' LIMIT 10) as tmp"

final_history_data = glueContext.read.format("jdbc")
             .option("driver", jdbc_driver_name)
             .option("url", db_url)
             .option("dbtable", query)
             .option("user", db_username)
             .option("password", db_password).load()

final_history = DynamicFrame.fromDF(final_history_data, glueContext, "final_history")

1. 最后将所有数据放入S3

glueContext.write_dynamic_frame.from_options(frame = final_history,
          connection_type = "s3",
          connection_options = {"path": "s3://glue-sample-target/output-dir/final_history"},
          format = "parquet")

哪种方法是最好的方法以及如何将jobBookmarkKeys应用于方法2？？

Answer 1

如果您使用需要处理的方法 1，您的工作可能会失败：

1. 请考虑没有为 dynamic_frame user_transaction 返回新数据，因为您启用了 jobbookmark，这将导致第 2 步为空。

2. 所以你需要有条件地检查它是否为空然后跳过处理并退出。

在方法 2 中，您不能利用 jobbookmark，因为 transformation_ctx 丢失并且不能包含在内，因为您通过将查询下推到引擎直接从数据库中读取。transformation_ctx 参数用于识别给定作业书签中的状态信息操作员。具体来说，AWS Glue 使用 transformation_ctx 来索引书签状态的键。更多信息请参考this。

比较这两种方法：

1. 如果在作业运行之间没有数据要处理，因为作业书签已启用，方法 1 将帮助您的作业运行得更快。

2. 如果您的数据库有一些好的硬件可以在更短的周转时间内返回查询结果，方法 2 将为您提供更好的性能。但是，如果在多个作业运行之间没有新数据，您的作业仍会将查询发送到数据库引擎，而在后一种情况下则不会。

3. 同时考虑到作业运行之间的新记录数，您可以决定选择作业书签还是将查询下推到数据库引擎。