Спасибо за обходной путь :) Я чувствую, что они должны это решить

я есть лямбда-процесс, который иногда опрашивает API для последних данных. Эти данные имеют уникальные ключи, и я хотел бы использовать Glue для обновления таблицы в MySQL. Есть ли возможность перезаписать данные с помощью этого ключа? (Аналогично режиму Spark = перезапись). Если нет - могу ли я обрезать таблицу в Glue перед вставкой всех новых данных?

Спасибо

 Thiago Ramos12 дек. 2017 г., 15:03
Мне интересно тот же вопрос.

Ответы на вопрос(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()
 hoaxz10 февр. 2018 г., 18:21
Как ты это делаешь в Клеевой Работе?
 Yuva04 июл. 2018 г., 18:55
@JoeC, я попытался использовать библиотеку pymysql в моей работе по склеиванию, но она не удалась из-за внешней библиотеки Служба поддержки AWS сообщает, что нам нужно использовать pymysql в качестве zip-файла и использовать его в качестве зависимого jar-файла. Это правда, или если нет, то можете ли вы помочь, как использовать / ссылку pymysql jar? Это работает в конечной точке разработки, но при запуске в качестве задания это не удается. Должно ли это быть добавлено как зависимая банка или контрольная банка? Техас
 JoeC14 февр. 2018 г., 15:54
@AlexMarkovDELETE FROM потребует перебора всех записей в клее, правильно? Где, как операции выше, все партии. Кроме того, есть некоторое (хотя и очень небольшое) количество времени, когда будут «отсутствующие» записи (что, в нашем случае, важно не иметь).
 JoeC13 февр. 2018 г., 23:49
@hoaxz см. правки для примера кода
 Alex Markov14 февр. 2018 г., 04:16
Мне любопытно, почему вы заменяете целевую БД из временной таблицы, а не просто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://...')
 JoeC18 дек. 2017 г., 17:05
Спасибо за обходной путь :) Я чувствую, что они должны это решить

Ваш ответ на вопрос