使用 AWS Glue 在两个 S3 存储桶之间加载数据时如何更新数据？

Question

这是我的第一个数据分析项目，我正在 AWS 上开发数据管道，管道步骤应如下所示：

1. 以 parquet 格式将数据从 RDS 导出到 S3（完成）。
2. 使用 Athena 查询 S3 中的数据（完成）。
3. 更新无效数据并将其传输到新的 S3 存储桶。
4. 从第二个S3存储桶中查询清理后的数据。

我陷入了步骤 3，在这一步中，验证团队必须通过从第一个 S3 存储桶查询数据、更新无效数据并将其复制到第二个 S3 存储桶来过滤有效数据并更新无效数据。

一个考虑因素是旧的无效数据必须按原样保留在第一个存储桶中，更新过程必须在传输过程中发生。

有没有办法在 Athena 中使用 UPDATE 语句？

我们可以在 AWS Glue 作业中传递所需的 UPDATE 语句吗？

我尝试使用此 问题 中的以下脚本来更新数据：

from pyspark.context import SparkContext
from awsglue.context import GlueContext
spark_session = glueContext.spark_session
sqlContext = SQLContext(spark_session.sparkContext, spark_session)
glueContext = GlueContext(SparkContext.getOrCreate())
dyF = glueContext.create_dynamic_frame.from_catalog(database='{{database}}',table_name='{{table_name}}'
df = dyF.toDF()
df.registerTempTable('{{name}}')
df = sqlContext.sql("{{sql_query_on_above_created_table}}")
df.format('parquet').save('{{s3_location}}')

但我得到了这个错误：

SyntaxError: invalid syntax (Untitled job.py, line 7)

我读到了有关将 AWS Data Pipeline 与 EMR 结合使用的信息，但我认为它很复杂，我无法想象它如何提供按需查询服务？

在两个 S3 存储桶之间传输数据并保持旧版本不变并将新数据放入新的 S3 存储桶中时更新数据的最佳解决方案是什么？

Answer 1

首先想到了第3步的几个选项

，雅典娜没有更新查询，但您可能会使用 ctas 查询

CREATE TABLE new_table
WITH (
      format = 'Parquet',
      external_location = 's3://my-other-bucket/'),
      write_compression = 'SNAPPY')
AS SELECT *, plusSomeTransformations
FROM existing_table;

或用胶水像您在问题中一样

from pyspark.context import SparkContext
from awsglue.context import GlueContext
spark_session = glueContext.spark_session
sqlContext = SQLContext(spark_session.sparkContext, spark_session)
glueContext = GlueContext(SparkContext.getOrCreate())
dyF = glueContext.create_dynamic_frame.from_catalog(database='myDB',table_name='existing_table'
df = dyF.toDF()
df.registerTempTable('myTable')
df = sqlContext.sql("SELECT *, plusSomeTransformations FROM myTable")
df.format('parquet').save('s3://my-other-bucket/')