AWS Glue: как ETL нескалярный JSON с различными схемами

Question

Цель

У меня есть папка S3, заполненная файлами json с различными схемами, включая массивы (резервное копирование DynamodB, как это происходит).Однако, хотя схемы различаются, все файлы содержат некоторые общие элементы, такие как «id» или «name», а также вложенные массивы различной длины, например «выбранные элементы».Я хочу быть в состоянии разобрать эти элементы на досуге.

У меня есть рабочий подход с использованием внешнего инструмента ETL (KNIME), который я собираюсь скопировать без использования сервера через Glue.

Фон

Рабочий подход:

1. Загрузка всех данных S3 в виде внешней таблицы с использованием Spectrum для каждой записи jsonкак одна varchar(65535) запись.
2. Разобрать нужные мне элементы, включая определенные массивы, используя такие функции Redshift SQL, как json_extract_path_text и json_array_length
3. Нормализовать требуемые массивы json путем перекрестного соединения с таблицей ссылок индексов массива
4. Выполните требуемое объединение таблиц и запишите их в Redshift для использования в Tableau

Теперь это похоже на задачу, которая должна соответствовать Glue.А именно, что я хочу сделать, это:

1. Загрузить все данные, используя Spectrum, как указано выше

2. В Glue создайте динамический фреймиз таблицы Spectrum

3. Анализировать данные из динамического фрейма, как указано выше, используя такие функции, как pyspark.sql.functions.explode() или, возможно, используя преобразование реляционного преобразования Glue

Или:

1. Скопируйте все скалярные данные в одну схему Glue (учитывая, что Glue еще не поддерживает массивы JSON)
2. Анализируйте JSON и разбивайте массивы, используя один из подходов, приведенных выше

Результаты пока

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

1. Сканирование данных JSON с использованием Glue - согласно этому посту , эвристика синтаксического анализатора Glue решает, что различные схемы источника слишком разныеотноситься к одному источнику и разбирать их как кучу разных таблиц.Было бы достаточно иметь сканер, который бы просто сканировал каждый файл, чтобы создать таблицу с одним столбцом типа varchar (65535), содержащим запись json на строку, но, похоже, нет никакого выражения пути JSON классификатора Glue, которое будетдостичь этого.

2. Подход «столбец с одним столбцом (65535)» может быть достигнут путем загрузки данных в виде внешней таблицы в Spectrum, однако кажется, что таблицы Spectrum не могут быть загружены в Glue как динамические кадры (примечаниечто соответствующая таблица существует в каталоге Glue, где она показана как имеющая ожидаемую схему varchar (65535).Работая в Zeppelin Notebooks, я обнаружил, что

   newFrame = glueContext.create_dynamic_frame.from_catalog(database="<spectrum database>", table_name="<spectrum one column varchar table>")

   Работает успешно, однако создает таблицу с newFrame.count() = 0 и newFrame.toDF().show(n), при любом значении n выдает нечетный выводв форме:

   ++ || ++ ++

   Короче говоря, pySpark не может работать напрямую с таблицами Spectrum через Glue.

3. Сканирование таблицы Spectrum с использованием Crawler.Здесь я указал сканеру на нужную таблицу через соединение Glue с моим кластером Redshift.Однако это дает сбой проверки конечной точки S3, который мне еще предстоит устранить.Я не хочу углубляться в конфигурацию VPC, учитывая, что я уже довольно сомневаюсь, что наведение Crawler за столом Spectrum будет правильным подходом.

Короче говоря,Я не нашел способа динамически принимать и анализировать нескалярный json на S3, используя Glue Crawler или комбинацию Glue и Redshift Spectrum.Возможно, это не эзотерическая задача - на самом деле это то, что должно быть достигнуто любым, кто хочет относительно автоматизированный подход к составлению отчетов о данных из веб-приложения на основе DynamodB.

Вопрос

Так что мой вопрос, в одном утверждении: есть ли способ проанализировать нескалярные файлы json на S3 с несовместимыми схемами, используя Glue (плюс, возможно, другой сервис AWS, такой как RS Spectrum)?

Answer 1

Итак, я предполагаю, что на заднем плане происходит несколько вещей.

Я полагаю, вы определили внешнюю таблицу в Redshift Spectrum, которая указывает на S3? Если так, то это не лучший подход. Вместо этого определите внешнюю схему, которая указывает на таблицу в каталоге данных Glue. В результате RS Spectrum увидит все в этой базе данных Glue, и вам не нужно определять отдельные таблицы.

Во-вторых, вы пытались определить таблицу вручную в каталоге данных Glue? Я провел обширное тестирование с форматом файлов Parquet, возиться с определениями таблиц и содержимым файлов. В результате все запросы из этой таблицы будут возвращать только данные, определенные в таблице. Таким образом, вы можете определить таблицу с полями 'id', 'name' и 'selected items', а все остальное будет игнорироваться.

Если по какой-либо причине предыдущая версия не работает, то поможет клейкая работа. Как примечание - всегда используйте только искру, никогда не используйте ничего связанного с glue_context. В любом случае, в spark.read.blaah вы можете указать параметр схемы. Используйте это.

dataSchema = StructType([StructField("id",StringType(),True) ,StructField("name",StringType(),True) ,StructField("selected items",ArrayType(.. etc ...),True) ])

Таким образом, вы получите тот же результат - он будет анализировать только те данные, которые вам нужны.

Настройка в конце, вероятно, должна выглядеть примерно так:

• S3 содержит JSON-файлы
• Задание склеивания считывает данные из входной группы и записывает их в другую область, возможно, разбитую на части и / или в другой формат данных
• Таблица склеивания определяется поверх второго ведра / префикса
• Redshift Spectrum указывает на базу данных, которая содержит таблицу, определенную на предыдущем шаге