通过AWS Glue将RDS Aurora Postgres数据推送至SQS的实现问题

现有代码的问题列表 #

• 数据源连接类型拼写错误：connection_type参数值写为postgreasql，正确值应为postgresql，会直接导致拉取RDS数据失败。
• job.commit()位置错误：该方法是Glue作业完成所有业务逻辑后才需要调用的终止方法，放在遍历发送SQS的逻辑之前，会导致后续代码完全不执行，作业提前结束。
• 无法直接遍历DynamicFrame：DynamicFrame是分布式数据集，直接在Driver端用for record in df遍历属于错误用法，无法正常获取存储在各Worker节点的分布式数据，还会引发严重性能问题。
• SQS客户端初始化位置错误：在Driver端初始化的boto3 SQS客户端无法序列化后传递到Spark Executor节点执行，会抛出序列化错误或连接异常。
• 消息格式错误：直接把DynamicRecord对象作为MessageBody传入SQS会报错，需要先将记录序列化为字符串格式（通常为JSON）。

修正后的代码 #

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

sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
logger = glueContext.get_logger()

# 拉取Postgres数据
df = glueContext.create_dynamic_frame.from_options(
    catalog_connection = "ams-connection", 
    connection_type="postgresql",
    connection_options={
        "dbtable": "input",
        "database": "ams"
    }
)

# 分布式处理每条记录发送SQS
def send_records_to_sqs(partition):
    # 每个Executor分区单独初始化SQS客户端，避免序列化问题
    sqs = boto3.resource('sqs', region_name='us-east-1')
    sqs_queue_url = "https://sqs.us-east-1.amazonaws.com/ACCID/SampleQue"
    queue = sqs.Queue(sqs_queue_url)
    
    for record in partition:
        try:
            # 将DynamicRecord转为字典再序列化为JSON字符串
            record_str = json.dumps(record.asDict())
            queue.send_message(MessageBody=record_str)
        except Exception as e:
            logger.error(f"发送消息失败: {str(e)}, 记录内容: {record.asDict()}")

# 调用分区处理逻辑
df.toDF().rdd.foreachPartition(send_records_to_sqs)

# 所有逻辑执行完成再提交作业
job.commit()

额外注意事项 #

• 确保Glue作业关联的IAM角色拥有SQS的sqs:SendMessage权限，以及Aurora Postgres的访问权限。
• SQS单条消息最大支持256KB，如果单条记录序列化后超过该阈值，需要先做截断、压缩或者改用S3中转再发送SQS通知。
• 如果数据量较大可以改用SQS批量发送接口，每次最多发送10条消息，大幅提升发送效率。

内容的提问来源于stack exchange，提问作者Aparup