Спасибо за обходной путь :) Я чувствую, что они должны это решить
я есть лямбда-процесс, который иногда опрашивает API для последних данных. Эти данные имеют уникальные ключи, и я хотел бы использовать Glue для обновления таблицы в MySQL. Есть ли возможность перезаписать данные с помощью этого ключа? (Аналогично режиму Spark = перезапись). Если нет - могу ли я обрезать таблицу в Glue перед вставкой всех новых данных?
Спасибо
Ответы на вопрос(2)
который я предложил, который немного проще, чем опубликованная альтернатива, заключается в следующем:
Создайте промежуточную таблицу в MySQL и загрузите новые данные в эту таблицу.Запустите команду: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 потребует перебора всех записей в клее, правильно? Где, как операции выше, все партии. Кроме того, есть некоторое (хотя и очень небольшое) количество времени, когда будут «отсутствующие» записи (что, в нашем случае, важно не иметь).DELETE FROM target_table а затем запустить остальную часть работы с клеем?и лучшее решение, которое мы могли придумать, было создать класс 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 на S3 вместе с зависимостью MySQL Jar и сделайте свою работу зависимой от них. В вашем скрипте PySpark вы можете загрузить свой метод усечения с помощью:
java_import(glue_context._jvm, "com.my.glue.utils.mysql.MySQLTruncateClient")
truncate_client = glue_context._jvm.MySQLTruncateClient()
truncate_client.truncate('my_table', 'jdbc:mysql://...')