AWS Клей для Redshift: возможно ли заменить, обновить или удалить данные?

Ответ 1

Это было решение, которое я получил от поддержки AWS Glue:

Как вы знаете, хотя вы можете создавать первичные ключи, Redshift не обеспечивает уникальность. Поэтому, если вы повторяете задания Glue, тогда дубликаты строк могут быть вставлены. Некоторые из способов сохранения уникальности:

• Используйте промежуточную таблицу, чтобы вставить все строки, а затем выполнить upsert/merge [1] в основной таблице, это нужно сделать за пределами клея.

• Добавьте еще один столбец в таблицу красного смещения [1], например временную метку вставки, чтобы дублировать, но знать, какой из них пришел первым или последним, а затем удалить дубликат после этого, если вам нужно.

• Загрузите ранее вставленные данные в dataframe, а затем сравните данные для вставки, чтобы избежать вставки дубликатов [3]

[1] - http://docs.aws.amazon.com/redshift/latest/dg/c_best-practices-upsert.html и http://www.silota.com/blog/amazon-redshift-upsert-support-staging-table-replace-rows/

[2] - https://github.com/databricks/spark-redshift/issues/238

[3] - https://docs.databricks.com/spark/latest/faq/join-two-dataframes-duplicated-column.html

Ответ 2

Работа закладки являются ключом. Просто отредактируйте работу и включите "Закладки работы", и она не будет обрабатывать уже обработанные данные. Обратите внимание, что задание необходимо выполнить повторно один раз, прежде чем оно обнаружит, что ему не нужно повторно обрабатывать старые данные.

Для получения дополнительной информации см.: http://docs.aws.amazon.com/glue/latest/dg/monitor-continuations.html.

Название "закладка", на мой взгляд, немного надуманно. Я бы никогда не взглянул на это, если бы случайно не наткнулся на него во время моих поисков.

Ответ 3

Сегодня я протестировал и получил обходной путь для обновления/удаления из целевой таблицы с использованием соединения JDBC.

Я использовал как показано ниже

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

import pg8000
args = getResolvedOptions(sys.argv, [
    'JOB_NAME',
    'PW',
    'HOST',
    'USER',
    'DB'
])
# ...
# Create Spark & Glue context

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# ...
config_port = ****
conn = pg8000.connect(
    database=args['DB'], 
    user=args['USER'], 
    password=args['PW'],
    host=args['HOST'],
    port=config_port
)
query = "UPDATE table .....;"

cur = conn.cursor()
cur.execute(query)
conn.commit()
cur.close()



query1 = "DELETE  AAA FROM  AAA A, BBB B WHERE  A.id = B.id"

cur1 = conn.cursor()
cur1.execute(query1)
conn.commit()
cur1.close()
conn.close()

Ответ 4

Функция закладки закладок в клее должна делать трюк, как было предложено выше. Я использовал его успешно, когда мой источник S3. http://docs.aws.amazon.com/glue/latest/dg/monitor-continuations.html

Ответ 5

Невозможно, к сожалению, сейчас! У меня был аналогичный вариант использования и проверен с поддержкой AWS. Они попросили меня запустить сценарий s3 COPY или temp stage/delete/insert с автономной машины.

Итак, да, его невозможно!

Ответ 6

Согласно моему тестированию (с тем же сценарием), функциональность BOOKMARK не работает. Повторяющиеся данные вставляются при многократном запуске задания. Я решил эту проблему, ежедневно удаляя файлы из местоположения S3 (через лямбду) и применяя таблицы Staging & Target. данные будут вставлены/обновлены на основе соответствующих ключевых столбцов.

Ответ 7

Пожалуйста, проверьте этот ответ. Существует объяснение и пример кода, как вставить данные в Redshift с помощью промежуточной таблицы. Тот же подход можно использовать для запуска любых запросов SQL до или после того, как Glue записывает данные с использованием параметров preactions и postactions:

// Write data to staging table in Redshift
glueContext.getJDBCSink(
  catalogConnection = "redshift-glue-connections-test",
  options = JsonOptions(Map(
    "database" -> "conndb",
    "dbtable" -> staging,
    "overwrite" -> "true",
    "preactions" -> "<another SQL queries>",
    "postactions" -> "<some SQL queries>"
  )),
  redshiftTmpDir = tempDir,
  transformationContext = "redshift-output"
).writeDynamicFrame(datasetDf)

Ответ 8

Просто протестировал функциональность "Bookmark Bookmark" в Glue и понял, что она не предотвращает дублирование записей, когда одно и то же задание запускается более одного раза. Есть ли более простой и простой способ загрузить файлы из S3 в Redshift через Glue?