AWS Glue

本文介绍了AWS Glue - 在插入之前截断目标 postgres 表的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我正在尝试在插入之前截断 postgres 目标表，并且通常尝试利用已在 GLUE 中创建的连接来触发外部函数.

I am trying to truncate a postgres destination table prior to insert, and in general, trying to fire external functions utilizing the connections already created in GLUE.

有人能做到吗?

推荐答案

我已经尝试了 DROP/TRUNCATE 方案，但是无法使用 Glue 中已经创建的连接来实现，但是使用纯 Python PostgreSQL 驱动程序 pg8000.

I've tried the DROP/ TRUNCATE scenario, but have not been able to do it with connections already created in Glue, but with a pure Python PostgreSQL driver, pg8000.

从pypi下载pg8000的tar

在根文件夹中创建一个空的__init__.py

压缩内容&上传到 S3

在作业的Python lib path中引用zip文件

将数据库连接详细信息设置为作业参数(确保在所有键名前加上 --).勾选服务器端加密"框.

Download the tar of pg8000 from pypi

Create an empty __init__.py in the root folder

Zip up the contents & upload to S3

Reference the zip file in the Python lib path of the job

Set the DB connection details as job params (make sure to prepend all key names with --). Tick the "Server-side encryption" box.

然后你可以简单地创建一个连接并执行 SQL.

Then you can simply create a connection and execute SQL.

import sys from awsglue.utils import getResolvedOptions from pyspark.context import SparkContext from awsglue.context import GlueContext from awsglue.dynamicframe import DynamicFrame from awsglue.job import Job import pg8000 args = getResolvedOptions(sys.argv, [ 'JOB_NAME', 'PW', 'HOST', 'USER', 'DB' ]) # ... # Create Spark & Glue context job = Job(glueContext) job.init(args['JOB_NAME'], args) # ... config_port = 5432 conn = pg8000.connect( database=args['DB'], user=args['USER'], password=args['PW'], host=args['HOST'], port=config_port ) query = "TRUNCATE TABLE {0};".format(".".join([schema, table])) cur = conn.cursor() cur.execute(query) connmit() cur.close() conn.close()