如何用AWS Glue将S3文件夹JSON文件导入PostgreSQL表？实操求助

一、调整AWS Glue Studio配置解决现有问题 #

1. 添加S3触发器到Glue作业 #

Glue Studio可视化界面不直接支持创建触发器，需在AWS Glue控制台单独配置：

• 进入AWS Glue控制台 → 左侧导航栏选择「触发器」→ 点击「添加触发器」
• 触发器类型选「事件触发」，填写名称后关联目标S3桶
• 事件类型选择s3:ObjectCreated:*，设置前缀为指定文件夹路径，过滤新增文件
• 最后选择要触发的Glue作业，完成创建

2. 在数据流中添加完整文件名 #

可视化作业里可通过S3数据源节点直接启用该功能：

• 编辑S3数据源节点 → 切换到「输出设置」标签
• 勾选「包含文件名」选项，系统会自动在数据流中添加filename列，值为文件的完整S3路径
• 若需自定义列名，可切换到脚本模式，用Spark函数手动添加：

  from pyspark.sql.functions import input_file_name
  df = df.withColumn("filename", input_file_name())
  

3. 连接自有PostgreSQL数据库 #

无需依赖Glue Data Catalog，直接用JDBC目标节点或脚本模式实现：

• 可视化方式：添加「JDBC」目标节点 → 配置连接参数：
  • JDBC URL填写：jdbc:postgresql://<host>:<port>/<dbname>
  • 用户名、密码可直接输入（推荐用AWS Secrets Manager存储密码，避免明文）
  • 映射列：将数据流中的filename和原始JSON数据列对应到目标表字段
• 脚本模式：直接编写Spark JDBC写入逻辑：

  df.write \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://your-host:5432/your-db") \
    .option("dbtable", "destination_table") \
    .option("user", "your-user") \
    .option("password", "your-password") \
    .mode("append") \
    .save()
  

二、更轻量替代方案：AWS Lambda #

针对你的基础需求，Lambda比Glue更灵活、成本更低：

实现步骤 #

1. 创建Lambda函数，选择Python（或Node.js）运行时
2. 添加S3触发器：进入Lambda函数「配置」→「触发器」→ 添加S3触发器，选择目标桶，事件类型设为s3:ObjectCreated:*，指定文件夹前缀
3. 编写Lambda代码（以Python为例）：

   import boto3
   import psycopg2
   import os
   
   s3_client = boto3.client('s3')
   
   def lambda_handler(event, context):
       # 解析S3触发事件，获取文件信息
       s3_record = event['Records'][0]['s3']
       bucket_name = s3_record['bucket']['name']
       file_key = s3_record['object']['key']
       full_filename = f"s3://{bucket_name}/{file_key}"
   
       # 读取S3文件原始内容
       s3_response = s3_client.get_object(Bucket=bucket_name, Key=file_key)
       file_contents = s3_response['Body'].read().decode('utf-8')
   
       # 连接PostgreSQL并插入数据
       conn = psycopg2.connect(
           host=os.environ['PG_HOST'],
           database=os.environ['PG_DB'],
           user=os.environ['PG_USER'],
           password=os.environ['PG_PASSWORD'],
           port=os.environ['PG_PORT']
       )
       cursor = conn.cursor()
       insert_sql = 'INSERT INTO destination_table (filename, data) VALUES (%s, %s)'
       cursor.execute(insert_sql, (full_filename, file_contents))
       conn.commit()
   
       # 关闭连接
       cursor.close()
       conn.close()
       return {"statusCode": 200, "message": "Data inserted successfully"}
   

4. 配置Lambda环境变量：将PostgreSQL的主机、数据库名、用户名、密码、端口存入环境变量，避免硬编码
5. 配置权限与网络：
   • 给Lambda添加S3读取权限（s3:GetObject）
   • 若PostgreSQL在VPC内，将Lambda部署到同一VPC的子网，配置安全组允许访问5432端口

注意事项 #

• 目标PostgreSQL表的data字段建议设为TEXT或JSONB类型，适配原始JSON内容
• 可给filename字段添加唯一约束，避免重复插入相同文件
• Lambda执行时间默认3秒，若文件较大可调整超时时间（最长15分钟）

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