Клей создает дубликаты записей, как это исправить? - PullRequest
Новая
       24

Клей создает дубликаты записей, как это исправить?

0 голосов
/ 08 марта 2019

В настоящее время мы используем Glue (скрипты Python) для переноса данных из базы данных MySQL в базу данных RedShift.Вчера мы обнаружили проблему: некоторые записи являются дубликатами, эти записи имеют один и тот же первичный ключ, который используется в базе данных MySQL.Согласно нашим требованиям, все данные в базе данных RedShift должны быть такими же, как в базе данных MySQL.

Я пытался удалить таблицу RedShift перед миграцией, но не нашел способа для этого ...

Не могли бы вы помочь мне решить проблему? 

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)

datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "glue-db", table_name = "table", transformation_ctx = "datasource0")
applymapping0_1 = ApplyMapping.apply(frame = datasource0, mappings = [...], transformation_ctx = "applymapping0_1")
resolvechoice0_2 = ResolveChoice.apply(frame = applymapping0_1, choice = "make_cols", transformation_ctx = "resolvechoice0_2")
dropnullfields0_3 = DropNullFields.apply(frame = resolvechoice0_2, transformation_ctx = "dropnullfields0_3")
datasink0_4 = glueContext.write_dynamic_frame.from_jdbc_conf(frame = dropnullfields0_3, catalog_connection = "redshift-cluster", connection_options = {"dbtable": "table", "database": "database"}, redshift_tmp_dir = args["TempDir"], transformation_ctx = "datasink0_4")

Мое решение: 

datasink0_4 = glueContext.write_dynamic_frame.from_jdbc_conf(frame = dropnullfields0_3, catalog_connection = "redshift-cluster", connection_options = {"dbtable": "mytable", "database": "mydatabase", "preactions": "delete from public.mytable;"}

Ответы [ 2 ]

0 голосов
/ 09 марта 2019

Если ваша цель не состоит в том, чтобы иметь дубликаты в таблице назначения, вы можете использовать опцию postactions для приемника JBDC (см. этот ответ для получения более подробной информации). В основном это позволяет реализовать объединение Redshift с использованием промежуточной таблицы.

Для вашего случая это должно быть так (заменяет существующие записи): 

post_actions = (
         "DELETE FROM dest_table USING staging_table AS S WHERE dest_table.id = S.id;"
         "INSERT INTO dest_table (id,name) SELECT id,name FROM staging_table;"
         "DROP TABLE IF EXISTS staging_table"
    )
datasink0_4 = glueContext.write_dynamic_frame.from_jdbc_conf(frame = dropnullfields0_3, catalog_connection = "redshift-cluster", connection_options = {"dbtable": "staging_table", "database": "database", "overwrite" -> "true", "postactions" -> post_actions}, redshift_tmp_dir = args["TempDir"], transformation_ctx = "datasink0_4")
0 голосов
/ 08 марта 2019

Redshift не накладывает ограничения уникальных ключей

Если вы не можете гарантировать, что ваши исходные скрипты избегают дублирования, вам нужно будет выполнить обычное задание для дедупликации при красном смещении, 

delete from yourtable
where id in
(
select id
from yourtable
group by 1
having count(*) >1
)
;

Вы рассматривали DMS как альтернативу клею?Это может работать лучше для вас.

...