Есть ли способ записывать только «хорошие» записи в таблицу SQL Server и возвращать «плохие» записи, используя задание AWS Glue? - PullRequest
Новая
       7

Есть ли способ записывать только «хорошие» записи в таблицу SQL Server и возвращать «плохие» записи, используя задание AWS Glue?

0 голосов
/ 09 апреля 2019

Я пытаюсь написать задание Glue (PySpark), выполняет некоторые ETL и в конечном итоге записывает эти данные в таблицу в SQL Server (определенную в каталоге AWS Glue). При записи записей в таблицу SQL Server могут существовать ограничения (например, первичные ключи, внешние ключи, типы столбцов), которые препятствуют записи определенных записей (то есть «плохих» записей) в таблицу. Когда это происходит, задание Glue выдает ошибку, и задание не выполняется. Есть ли способ предотвратить провал всей работы? Вместо этого можно ли было бы писать только «хорошие» записи и возвращать «плохие» записи, которые нарушили SQL Server, обратно в задание Glue (чтобы их можно было загрузить на S3)?

Я использую функцию write_dynamic_frame_from_catalog для записи данных в таблицу SQL Server. Вот пример кода для контекста: 

# perform etl 
output_df=spark.sql("SELECT ...")

# create dataframe and write to SQL Server
output_dynamic_frame = DynamicFrame.fromDF(output_df, glueContext, 'output_dynamic_frame')
glueContext.write_dynamic_frame_from_catalog(frame = output_dynamic_frame, database="<DATABASE_NAME>", table_name="<TABLE_NAME>")

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

1 Ответ

0 голосов
/ 10 апреля 2019

Я думаю, вы можете использовать AWS Glue для извлечения данных из вашей БД в S3, а затем с помощью Pyspark вы можете получить «плохие записи» при чтении файлов S3: 

corruptDF = (spark.read   
            .option("mode", "PERMISSIVE")  
            .option("columnNameOfCorruptRecord", "_corrupt_record")  
            .csv("s3://bucket-name/path")

Затем вы можете отфильтроватьс помощью поля columnNameOfCorruptRecord и сохраните «хорошие» в вашей БД, а «плохие» - в путь S3.

Кроме того, в Databricks есть функция обработки плохих записей и файлов здесь, где вы можете указать опцию badRecordsPath при чтении файла, чтобы «записи мешков» отправлялись по этому пути.Имейте в виду, что это работает только при чтении CSV, JSON и любых встроенных источников на основе файлов (например, паркет)

...