Apache Воздушный поток - Как подойти к геокодированию? - PullRequest
Новая
       3

Apache Воздушный поток - Как подойти к геокодированию?

2 голосов
/ 28 апреля 2020

Пытаюсь изучить Apache Поток воздуха, но изо всех сил пытаюсь понять, как лучше всего подойти к одному конкретному аспекту моей DAG re: геокодирование с помощью внешнего API отдыха.

Предполагая, что DAG:

задача 1: загрузка данных - сохраняет в CSV.

задача 2: очистка, ETL и т. д. c - запись в базу данных.

Должен ли я тогда ...?

задача 3:

  • A) Подключается к базе данных,
  • B) извлекает первые N записей, для которых требуется геокодирование
  • C). запрос первой партии к внешнему API и ожидает ответа
  • D) записывает обратно в БД.
  • E) Промыть и повторять AD, пока все записи не будут геокодированы
  • F) Выходная задача

Является ли такая итеративная потенциально долго выполняемая задача в порядке, пока она выход после обработки всех записей или я принципиально неправильно понимаю / неправильно использую воздушный поток?

Какие-либо лучшие практики / примеры в publi c основах кода?

1 Ответ

1 голос
/ 28 апреля 2020

Я думаю, что такой подход хорош. Вы пытаетесь избежать использования XCOM, объединяя все этапы в один оператор, что, на мой взгляд, является достойной идеей и распространенным делом.

Что-то еще может сделать, это разделить задачу на несколько дополнительных шагов и «переставить» данные между шагами, что позволит вам иметь больше операторов и не использовать XCOM:

Оператор 1 (База данных -> Файловая система)

  • Подключение к базе данных
  • Запуск SQL для извлечения N записей из базы данных
  • Запись этих записей в файловую систему как CSV

Оператор 2 (API вызовов с записями CSV)

  • Считывание записей из файловой системы
  • API вызовов с записями
  • Запись новых записей в файловую систему как CSV

Оператор 3 (Файловая система -> База данных)

  • Чтение записей из CSV вне файловой системы
  • Запись записей в базу данных

Преимущество такого подхода состоит в том, что, скажем, в случае сбоя вызова API вам не нужно повторно запускать извлечение из базы данных. Вы просто перезапустите задачу API.

...