是否可以从 AWS Glue python 作业执行任意 SQL 命令，例如 ALTER TABLE？我知道我可以使用它从表中读取数据，但是有没有办法执行其他数据库特定命令？
我需要将数据提取到目标数据库中，然后立即运行一些 ALTER 命令。
因此，在进行了广泛的研究并在 AWS 支持下打开了一个案例后，他们告诉我目前无法通过 Python shell 或 Glue pyspark 作业进行操作。但我只是尝试了一些有创意的东西，它奏效了！这个想法是使用 Sparks 已经依赖的 py4j 并利用标准的 java sql 包。
这种方法的两大好处：
这样做的一个巨大好处是，您可以将数据库连接定义为 Glue 数据连接，并将 jdbc 详细信息和凭据保留在其中，而无需将它们硬编码在 Glue 代码中。我下面的示例通过调用
glueContext.extract_jdbc_conf('your_glue_data_connection_name')如果您需要在受支持的开箱即用的 Glue 数据库上运行 SQL 命令，您甚至不需要为该数据库使用/传递 jdbc 驱动程序 - 只需确保为该数据库设置 Glue 连接并添加该连接即可到您的 Glue 作业 - Glue 将上传正确的数据库驱动程序 jar。
请记住，下面的代码是由驱动程序进程执行的，不能由 Spark 工作线程/执行程序执行。
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
logger = glueContext.get_logger()
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
# dw-poc-dev spark test
source_jdbc_conf = glueContext.extract_jdbc_conf('your_glue_database_connection_name')
from py4j.java_gateway import java_import
java_import(sc._gateway.jvm,"java.sql.Connection")
java_import(sc._gateway.jvm,"java.sql.DatabaseMetaData")
java_import(sc._gateway.jvm,"java.sql.DriverManager")
java_import(sc._gateway.jvm,"java.sql.SQLException")
conn = sc._gateway.jvm.DriverManager.getConnection(source_jdbc_conf.get('url'), source_jdbc_conf.get('user'), source_jdbc_conf.get('password'))
print(conn.getMetaData().getDatabaseProductName())
# call stored procedure, in this case I call sp_start_job
cstmt = conn.prepareCall("{call dbo.sp_start_job(?)}");
cstmt.setString("job_name", "testjob");
results = cstmt.execute();
conn.close()
几个小时后我终于开始工作了，希望以下内容对您有所帮助。我的剧本深受早期回复的影响，谢谢。
先决条件：
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
glue_connection_name = '[Name of your glue connection (not the job name)]'
database_name = '[name of your postgreSQL database]'
stored_proc = '[Stored procedure call, for example public.mystoredproc()]'
#Below this point no changes should be necessary.
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
glue_job_name = args['JOB_NAME']
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(glue_job_name, args)
job.commit()
logger = glueContext.get_logger()
logger.info('Getting details for connection ' + glue_connection_name)
source_jdbc_conf = glueContext.extract_jdbc_conf(glue_connection_name)
from py4j.java_gateway import java_import
java_import(sc._gateway.jvm,"java.sql.Connection")
java_import(sc._gateway.jvm,"java.sql.DatabaseMetaData")
java_import(sc._gateway.jvm,"java.sql.DriverManager")
java_import(sc._gateway.jvm,"java.sql.SQLException")
conn = sc._gateway.jvm.DriverManager.getConnection(source_jdbc_conf.get('url') + '/' + database_name, source_jdbc_conf.get('user'), source_jdbc_conf.get('password'))
logger.info('Connected to ' + conn.getMetaData().getDatabaseProductName() + ', ' + source_jdbc_conf.get('url') + '/' + database_name)
stmt = conn.createStatement();
rs = stmt.executeUpdate('call ' + stored_proc);
logger.info("Finished")
我修改了米什金共享的代码，但它对我不起作用。因此，在进行了一些故障排除后，我意识到目录中的连接不起作用。所以我必须手动修改它并稍微调整代码。现在它可以工作，但最终会出现异常，因为它无法将 java 结果转换为 python 结果。我做了一个解决办法，所以请谨慎使用。
below is my code.
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
## @params: [TempDir, JOB_NAME]
args = getResolvedOptions(sys.argv, ['TempDir','JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
#source_jdbc_conf = glueContext.extract_jdbc_conf('redshift_publicschema')
from py4j.java_gateway import java_import
java_import(sc._gateway.jvm,"java.sql.Connection")
java_import(sc._gateway.jvm,"java.sql.DatabaseMetaData")
java_import(sc._gateway.jvm,"java.sql.DriverManager")
java_import(sc._gateway.jvm,"java.sql.SQLException")
print('Trying to connect to DB')
conn = sc._gateway.jvm.DriverManager.getConnection('jdbc:redshift://redshift-cluster-2-url:4000/databasename', 'myusername', 'mypassword')
print('Trying to connect to DB success!')
print(conn.getMetaData().getDatabaseProductName())
# call stored procedure, in this case I call sp_start_job
stmt = conn.createStatement();
#cstmt = conn.prepareCall("call dbname.schemaname.my_storedproc();");
print('Call to proc trying ')
#cstmt.setString("job_name", "testjob");
try:
rs = stmt.executeQuery('call mySchemaName.my_storedproc()');
except:
print("An exception occurred but proc has run")
#results = cstmt.execute();`enter code here`
conn.close()
如果将连接对象附加到粘合作业，您可以轻松获取连接设置：
glue_client = boto3.client('glue')
getjob=glue_client.get_job(JobName=args["JOB_NAME"])
connection_settings = glue_client.get_connection(Name=getjob['Job']['Connections']['Connections'][0])
conn_name = connection_settings['Connection']['Name']
df = glueContext.extract_jdbc_conf(conn_name)
这要看情况。如果您使用 redshift 作为目标，则可以选择指定前操作和后操作作为连接选项的一部分。您可以在那里指定更改操作。然而，对于其余的目标类型，您可能需要使用一些 python 模块，例如 pg8000（在 Postgres 的情况下）和其他
感谢米什金分享脚本。当我按照 Redshift 脚本操作时，出现以下错误
调用 z:java.sql.DriverManager.getConnection 时发生错误。 [亚马逊]需要 JDSI 连接设置中不存在设置 ConnSchema
看起来
source_jdbc_conf.get('url')是的，原来的帖子也对我有用，但是当你调用存储过程时，你确实需要包含数据库名称，而且你确实需要对连接上的 URL 进行硬编码，因为这对我来说是错误的......
这对雪花有用吗？ Snowflake 的 JDBC url 格式看起来不同并且不符合脚本需求？
Error Category: UNSUPPORTED_OPERATION_ERROR; An error occurred while calling o99.extractJDBCConf. JDBC connection URL jdbc:snowflake://XXXXXXX.us-east-1.snowflakecomputing.com/?warehouse=XXXXXX&db=XXXXX_DEV&schema=XXXXX is not supported. Check the Developer Guide for the list of supported data stores / URL formatting.