我有一个 lambda 进程，它偶尔会轮询 API 以获取最近的数据。该数据具有唯一键，我想使用 Glue 更新 MySQL 中的表。是否有使用此密钥覆盖数据的选项？ （类似于 Spark 的模式=覆盖）。如果没有 - 我是否可以在插入所有新数据之前在 Glue 中截断表格？
谢谢
【问题讨论】：
标签： mysql amazon-web-services pyspark aws-glue
我发现了一种在 Glue 中使用 JDBC 连接的更简单方法。当您将数据写入 Redshift 集群时，Glue 团队建议截断表的方式是通过以下示例代码：
datasink5 = glueContext.write_dynamic_frame.from_jdbc_conf(frame = resolvechoice4, catalog_connection = "<connection-name>", connection_options = {"dbtable": "<target-table>", "database": "testdb", "preactions":"TRUNCATE TABLE <table-name>"}, redshift_tmp_dir = args["TempDir"], transformation_ctx = "datasink5")
在哪里
connection-name your Glue connection name to your Redshift Cluster
target-table the table you're loading the data in
testdb name of the database
table-name name of the table to truncate (ideally the table you're loading into)
【讨论】：
我想出的解决方法比发布的替代方法简单一些，如下所示：
REPLACE INTO myTable SELECT * FROM myStagingTable;
这可以通过：
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: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
import pymysql
pymysql.install_as_MySQLdb()
import MySQLdb
db = MySQLdb.connect("URL", "USERNAME", "PASSWORD", "DATABASE")
cursor = db.cursor()
cursor.execute("REPLACE INTO myTable SELECT * FROM myStagingTable")
cursor.fetchall()
db.close()
job.commit()
【讨论】：
DELETE FROM target_table 然后运行其余的 Glue 作业？
DELETE FROM 需要遍历胶水中的所有记录，对吗？上面的操作都是批处理的。此外，有一些（尽管非常少）时间会出现“丢失”记录（在我们的例子中，重要的是不要有）。
我在使用 Redshift 时遇到了同样的问题，我们能想到的最佳解决方案是创建一个 Java 类来加载 MySQL 驱动程序并发出一个截断表：
package com.my.glue.utils.mysql;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
@SuppressWarnings("unused")
public class MySQLTruncateClient {
public void truncate(String tableName, String url) throws SQLException, ClassNotFoundException {
Class.forName("com.mysql.jdbc.Driver");
try (Connection mysqlConnection = DriverManager.getConnection(url);
Statement statement = mysqlConnection.createStatement()) {
statement.execute(String.format("TRUNCATE TABLE %s", tableName));
}
}
}
将该 JAR 连同您的 MySQL Jar 依赖项一起上传到 S3，并使您的工作依赖于这些。在您的 PySpark 脚本中，您可以使用以下命令加载 truncate 方法：
java_import(glue_context._jvm, "com.my.glue.utils.mysql.MySQLTruncateClient")
truncate_client = glue_context._jvm.MySQLTruncateClient()
truncate_client.truncate('my_table', 'jdbc:mysql://...')
【讨论】：