Как использовать воздушный поток для оркестровки простых скриптов Python etl?

Question

Мне нравится идея воздушного потока, но я застрял в основах.Со вчерашнего дня у меня работает airflow на решении vm ubuntu-postgres.Я вижу панель инструментов и данные примера :)) Теперь я хочу перенести пример сценария, который я использую для обработки необработанных данных в подготовленные.

Представьте, что у вас есть папка csv-файлов.Сегодня мой скрипт проходит через него, передавая каждый файл в список, который будет преобразован в df.После этого я готовлю имена их столбцов, делаю некоторую очистку данных и записываю их в другой формат.

1: pd.read_csv для файлов в каталоге

2: создать df

3: очистить имена столбцов

4: очистить значения (параллельноSTP 3)

5: записать результат в базу данных

Как мне организовать свои файлы в соответствии с потоком воздуха?Как должен выглядеть скрипт?Я передаю один метод, один файл или мне нужно создать несколько файлов для каждой части?На данный момент мне не хватает основной концепции :( Все, что я читаю об воздушном потоке, намного сложнее, чем мой простой случай. Я подумывал также отойти от воздушного потока до Бонобо, Мары, Луиджи, но я думаю, что воздушный поток того стоит?!

Answer 1

Я бы использовал PythonOperator, поместил весь код в функцию Python, создал бы одну задачу Airflow и все.

Можно также добавить загрузку файлов csv в функцию и запись в базу данных, если необходимо разделить эти шаги.Все это будет помещено в один DAG.

Таким образом, ваш DAG будет иметь три задачи, например:

loadCSV (PythonOperator)
parseDF (PythonOperator)
pushToDB (PythonOperator)

Если вы используете несколько задач, вам нужно , чтобы использовать Воздушный поток XCom .В начале проще всего использовать одну задачу.

Здесь есть несколько примеров кода под тегом airflow.Когда вы что-то создали, спросите еще раз.