Вопрос о плане миграции сложного конвейера данных

Question

Мой план:

1. Переместить всю обработку данных в Spark (желательно PySpark), чтобы окончательный вывод данных (для потребителей) шел только в Redshift.Похоже, что Spark хорошо подключается ко всем различным источникам (Dynamo DB, S3, Redshift).Вывод на Redshift / S3 и т. Д. В зависимости от потребностей клиента.Это позволяет избежать нескольких кластеров Redshift, поломок / чрезмерного использования внутренних неподдерживаемых инструментов ETL, копирования одних и тех же данных в кластеры, представления и таблицы и т. Д. (Что является текущей настройкой).
2. Используйте Luigi для создания веб-интерфейса пользователя для ежедневного мониторинга конвейеров, визуализации дерева зависимостей и планирования ETL.Уведомления по электронной почте должны быть опцией для сбоев также.Альтернативой является конвейер данных AWS, но у Луиджи, похоже, лучший интерфейс для того, что происходит, когда задействовано много зависимостей (некоторые деревья имеют 5 уровней глубины, но, возможно, этого также можно избежать с помощью лучшего кода Spark).

Вопросы:

1. Интегрируется ли Луиджи с Spark (раньше я использовал только PySpark, а не Луиджи, поэтому для меня это кривая обучения).План состоял в том, чтобы запланировать «приложения», и Spark на самом деле тоже имеет ETL, так что я не уверен, как Луиджи интегрируется здесь?
2. Как объяснить тот факт, что некоторые конвейеры могут работать в режиме реального времени - нужно ли мне, например, ежечасно раскручивать задание Spark / EMR?

Яоткрыты для мыслей / предложений / лучших способов сделать это тоже!

Answer 1

Чтобы ответить на ваши вопросы напрямую,

1) Да, Луиджи отлично играет с PySpark, как и любая другая библиотека.Мы, безусловно, запускаем его без проблем - единственное предостережение в том, что вам нужно быть немного осторожнее с import s и иметь их в функциях класса Luigi, поскольку в фоновом режиме он раскручивает новые экземпляры Python.

2) Есть способы заставить Луиджи хлебать в потоках данных, но это сложно сделать.Реально, вы бы вернулись к выполнению ежечасного цикла cron, чтобы просто вызвать конвейер и обработать новые данные.Этот вид отражает сценарий использования Spotify для Luigi, где они ежедневно выполняют задания для расчета топ-артиста и т. Д.

Как @RonD предлагает, если бы я сейчас строил новый конвейер, я бы пропустил Luigi и сразу пошел бы кРасход воздуха.Если ничего другого, посмотрите историю выпусков.Луиджи действительно долго не работал (потому что он работает для главного разработчика).Принимая во внимание, что AirFlow активно инкубируется Apache.

Answer 2

Вместо Luigi используйте Apache Airflow для оркестровки рабочего процесса (код написан на Python).Он имеет множество операторов и встроенных хуков, которые вы можете вызывать в DAG (Workflows).Например, создайте задачу для вызова оператора для запуска кластера EMR, другую для запуска сценария PySpark, расположенного в s3 на кластере, другую для отслеживания выполнения для состояния.Вы также можете использовать задачи для настройки зависимостей и т. Д.