Почему такое внимание к воздуху?
Процентная ставка воздушного потока. Источник: Google Trends
«Программная индустрия наблюдала заметное повышение в разработке инструментов для управления данными. Эти инструменты варьируются от решений для хранения, которые содержат данные о домашних условиях для трубопроводов, которые транспортируют данные ».
Компании, управляемые данными, такие как Airbnb, Quizlet полагаются на эти данные, чтобы разрешить некоторые утомительные задачи, такие как
- Планирование
- Тестирование
- Обработка ошибок
- Версию
- Масштабирование
текущий поток данных настроен.
Мы называем эти трубопроводы данных. И известный инструмент в этом пространстве это Apache Airflow Отказ
Воздушный поток решает эти утомительные задачи,
Нажмите на функцию, если вы хотите перейти в свой раздел напрямую.
И лучшая часть всего AirFlow – это инструмент с открытым исходным кодом и имеет быстро растущую базу пользователя и вклад 🙂
Без дальнейших церемоний,
Давайте глубоким погружением в рамки рабочего процесса
И создать минималистичный трубопровод для рабочего процесса, который использует все функции воздушных потоков, которые я перечисленных выше.
✨ dags.
- Конфигурация DAG –
Default_args. - Экземпляр DAG
Dag () - DAG задачи –
Операторы
Конфигурация DAG – default_args
Default_args используется для установки свойств (Аргументы) которые распространены для всех задач.
# Initializing the default arguments that we'll pass to our DAG
default_args = {
'owner': 'Srinidhi',
'start_date': datetime(2018, 1, 1),
'retries': 1,
'on_failure_callback': task_failure_alert
}
Экземпляр DAG – DAG ()
# Creating a DAG instance
my_workflow_dag = DAG(
'first_dag',
default_args=default_args,
description='Sample DAG for DE Series',
schedule_interval='0 0 * * *',
)
DAG задачи – операторы
Операторы по воздушным потокам используются при создании отдельных задач в DAG. Свойства, специфичные для задачи, будут настроены в операторе.
Некоторые из наиболее часто используемых операторов воздушных потоков
Pythonoperator – вызывает функцию Python. Альтернативный способ создания задачи функции Python – использовать API Taskflow, который доступен от AirFlow 2.0 Башоператор – Выполняет команду UNIX
#Operators
# Bash operator tasks that executes unix command - echo
task_1 = BashOperator(
task_id='first_task',
bash_command='echo "First Task"',
dag=my_workflow_dag,
)
# Python Operator that prints the details of current job using context variable
task_2 = PythonOperator(
task_id='second_task',
python_callable=print_fn,
provide_context=True,
on_success_callback = task_success_alert,
dag=my_workflow_dag,
)
#Python Function
def print_fn(ds):
"""
Implement your python task logic here.
"""
#Prints job details passed through provide_context parameter
print(ds)
Теперь, когда у нас есть наш даг готов, давайте посмотрим, где я настроил в нем функции.
✨ повторить попытку
✨ оповещения
👇 Образец слабые уведомления, полученные от воздушного потока
Просмотр кода – успех, неудача
✨ Webui
Поскольку воздушный поток изготовлен из Flask Framework, вы можете даже расширить Webui, создав дополнительные страницы, используя Флэк AppBuilder Отказ
✨ планирование
В дополнение к этому, DAGS может быть сделан для автоматического запуска в определенное время, предоставляя Обозначение Cron в SCARTE_INTERVAL
✨ API
Настройка экземпляра воздушного потока
Вот несколько отличных ресурсов: В окнах с докером На Ubuntu Local Setup На Кубернес AWS использует облачность
Если у вас нет пропускной способности, чтобы настроить и поддерживать воздушный поток в своей собственной инфраструктуре, вот коммерческие поставщики AirFlow-As-A-A-Service:
Google Cloud Composer – Управляемый сервис, построенный в Google Cloud и Airflow. Astronomer.io – В дополнение к хостингу воздушного потока в их инфраструктуре они предоставляют решения, ориентированные на услуги по воздушным потокам и поддержке.
Это воздушный поток единственный вариант для рабочих процессов?
Конечно, не Отказ Это зависит от разных факторов. Скажите, что весь ваш продукт размещен в одном провайдере облака (AWS/Azure/GCP), и вы в порядке с блокировкой поставщика. Затем,
- Для AWS, AWS Шаг Функция будет хороший вариант.
- Для Azure вы можете выбрать Фабрика данных Azure Factory Отказ
- Для GCP, Google Cloud Composer будет лучшим в форме.
Принимая во внимание, что если вы хотите провести свой продукт в мульти-облаке – Воздушный поток будет лучше подходит. Есть и другие рабочие процессы на рынке, похожие на AirFlow: Префект Луиджи Дагстер
Лучшие практики/советы по воздушным потокам:
- Как только у вас установлен воздушный поток, измените конфигурации вашего экземпляра воздушного потока в файле Master Config – airflow.cfg.
- Избегайте создания сложной логики обработки данных в воздушном потоке, так как воздушный поток не предназначен для работы в качестве механизма обработки данных. Основное внимание уделяется оркетеру задания потока данных. (Для обработки данных запустите сценарий обработки в пакетной вычислительной системе или Spark Framework и вызовите из воздушного потока)
- Использовать Макросы и шаблоны Чтобы избежать жестких значений кодирования.
- Во время миграции метаданных воздушных потоков DB в новую базу данных, используйте “Airflow Upgradedb” (для обновления воздушного потока 2,0 дБ) вместо «initddb initdb» (для init init init init inite)
- Когда вы хотите реализовать задачу из коробки, создайте Пользовательские операторы используя
БезопасностьКласс для повторного использования их. - Использовать Плагины воздушного потока , подключиться к сторонним инструментам/источникам данных.
- Используйте базу данных Postgres в качестве метаданных по воздушным потокам DB для использования локальных и цеменчиков.
Если у вас есть какие-либо вопросы или все больше советы, которые, по вашему мнению, могут быть полезны, комментарий ниже на пост.
Сдвиньте вперед, я буду публиковать подробные посты на инструменты и каркасы, используемые день инженеров данных и выходным.
Следуйте за обновлениями.
Оригинал: “https://dev.to/srinidhi/data-engineering-series-3-apache-airflow-the-modern-workflow-management-tool-what-do-you-need-to-know-78l”