Прежде чем перейти к статье, хочу вам представить, экономическую онлайн игру Brave Knights, в которой вы можете играть и зарабатывать. Регистируйтесь, играйте и зарабатывайте!
Мы познакомим вас с новыми функциями KubernetesExecutor 2.0. Внимание, спойлер!!! Процесс стал быстрее, гибче и проще для понимания.
Вместе с Airflow 2.0 мы рады представить полностью переработанный KubernetesExecutor. Эта новая архитектура быстрее, гибче и проще для понимания, чем KubernetesExecutor 1.10. В качестве первого шага мы хотели бы познакомить вас с новыми функциями KubernetesExecutor 2.0!
Что такое KubernetesExecutor?
В 2018 году мы представили KubernetesExecutor, основанный на идеях автомасштабирования и гибкости. В Airflow еще не было четкой концепции автомасштабирования Celery Workers (хотя наша недавняя работа с KEDA в этом отношении достигла больших успехов), поэтому мы хотели создать систему, которая могла бы соответствовать потребностям пользователя. В результате этого исследования была создана система, использующая Kubernetes API для запуска задачи pod per-airflow. Ценным побочным эффектом этой системы на основе Kubernetes API является то, что она открыла возможность для пользователей добавлять уникальные дополнения и ограничения для каждой задачи.
Используя Kubernetes API и KubernetesExecutor, пользователи Airflow могут определить, что определенные задачи имеют доступ к определенным секретам или что задача может выполняться только на узле, существующем в Европейском Союзе (что может быть полезно для управления данными). Пользователи также могут указать, сколько ресурсов занимает задача, что может сильно различаться в зависимости от того, что делает задача (например, для запуска сценария TensorFlow потребуется доступ к графическим процессорам). С помощью этого API KubernetesExecutor позволяет инженерам данных иметь гораздо более точный контроль над тем, как Airflow выполняет свои задачи, чем они использовали бы только существующие очереди Celery.
Конечно, KubernetesExecutor не оптимален для всех случаев использования. Поскольку он запускает pod для каждой задачи, системные издержки могут замедлить выполнение заданий, чем существующие рабочие Celery с горячим запуском (хотя это порядка секунд, что несущественно для длительных задач). Кроме того, CeleryExecutor будет более эффективным при сверхвысоком объеме, поскольку он может выполнять несколько задач на одном рабочем месте. Учитывая, что и CeleryExecutor, и KubernetesExecutor имеют уникальные преимущества для пользователей Airflow, одной замечательной особенностью Airflow 2.0 является то, что теперь у нас есть CeleryKubernetesExecutor, который позволяет пользователям использовать преимущества обеих систем!
Новые возможности KubernetesExecutor
Файл podtemplate
В Airflow 1.10.12 мы представили файл pod_template_file. Этот новый способ хранения всех конфигураций Kubernetes включал переписывание внутренних компонентов KubernetesExecutor. Однако это изменение того стоило, поскольку администраторы Airflow теперь могут использовать весь API Kubernetes при создании шаблонов для своих инженеров по данным.
Это изменение также открывает дверь к возможности поддерживать несколько pod_template_files в будущих выпусках Airflow. Пользователи смогут выбрать pod_template_file, который лучше всего соответствует их варианту использования, почти так же, как пользователи CeleryExecutor выбирают отдельные очереди.
Эта функция запуска pod на основе pod_template_files, объединенных с добавлением 2.0 быстрого выполнения задачи, означает, что мы можем запускать pod в Kubernetes, которые не должны завершаться после выполнения одной задачи. Вместо этого эти pod могут выполнять новые задачи во многом так же, как и рабочие Celery. Результат — значительное ускорение выполнения задачи KubernetesExecutor.
Execitor_config
Airflow 2.0 предлагает новый файл executor_config, который значительно более гибкий для пользователя. Вместо того, чтобы ограничиваться словарем Python с ограниченным доступом к функциям, пользователи теперь могут использовать весь API Kubernetes. Мы решили изменить значение ключа словаря executor_config на podOverride. Мы одновременно подумали, что этот ключ является более описательным и оставляет открытой возможность добавления дополнительных опций в будущем.
Стоит отметить, что устаревшие значения executeor_config по-прежнему будут работать в Airflow 2.0, чтобы минимизировать критические изменения для миграции. Однако они будут исключены и будут удалены в будущей версии, поэтому мы рекомендуем как можно скорее переключиться на них.
podmutationhook
Как было введено в 1.10.12, новый pod_mutation_hook принимает объект Kubernetes V1Pod в качестве параметра и позволяет администратору Airflow изменять все pod с помощью Kubernetes API до того, как Airflow выпустит эти pod. Этот хук применяется как к pod, созданным KubernetesExecutor, так и к pod, созданным KubernetesPodOperator.
Упрощенный дизайн
Мы рассмотрели три основные функции KubernetesExecutor. Мы видели, как pod_template_file обеспечивает полную гибкость для создания начальных шаблонов pod, как Kubernetes pod_override позволяет пользователям изменять задачи на лету и как pod_mutation_hook дает администратору переопределения перед релизом pod. Однако больше всего меня восхищают эти новые функции, как красиво они работают в упрощенном дизайне.
Ниже мы можем увидеть схему того, как раньше работала архитектура KubernetesExecutor.
Оглядываясь назад, можно сказать, что это похоже на фабрику, которая постепенно создает контейнер перед запуском. Pod проходит как минимум четыре этапа и три класса, прежде чем он будет сериализован и запущен. Такой дизайн оставляет много места для ошибок и недоразумений.
Сравним это с дизайном.
Этот новый дизайн намного проще. Теперь мы просто берем существующую спецификацию pod, позволяя один раунд переопределений пользователя и раунд переопределений администратора перед запуском. Все три пользователя изменяют один и тот же объект V1pod, что приводит к большей согласованности во всей организации.
Что это значит для инженеров по обработке данных
Все описанные выше функции явно важны для администраторов Airflow или инженеров DevOps, но некоторые из этих новых функций также предназначены для непосредственного использования инженерами по обработке данных.
Самым большим изменением для инженера по обработке данных, который пишет группы DAG, вероятно, является новый файл executor_config с podOverride. До этого изменения, если инженер хотел переопределить системные настройки Kubernetes по умолчанию для определенных задач в своих DAG, они были ограничены в том, какие функции они могли получить без необходимости использовать KubernetesPodOperator для этого. Использование KubernetesPodOperator включало создание и построение всего образа Docker для запуска каждой задачи, что могло добавить сложности и усилий. Теперь, с новым executeor_config, инженеры могут получить доступ ко всему Kubernetes API и использовать podOverride для таких вещей, как монтирование томов, добавление переменных среды, добавление меток, установка сходства узлов и т. Д. С помощью любого оператора, который они пожелают.
Возьмем, к примеру, случай, когда мы хотим добавить метку и переменную среды, которая будет использоваться в функции Python для pod, выполняющего нашу задачу. С новым executeor_config мы действительно можем передать этот podOverride очень чисто, используя PythonOperator с новым API TaskFlow. Этот DAG будет выглядеть примерно так:
from airflow.decorators import dag, task
from datetime import datetime
import os
import json
import requests
from kubernetes.client import models as k8s
new_config ={ "pod_override": k8s.V1Pod(
metadata=k8s.V1ObjectMeta(labels={"purpose": "pod-override-example"}),
spec=k8s.V1PodSpec(
containers=[
k8s.V1Container(
name="base",
env=[
k8s.V1EnvVar(name="STATE", value="wa")
],
)
]
)
)
}
default_args = {
'start_date': datetime(2021, 1, 1)
}
@dag('k8s_executor_example', schedule_interval='@daily', default_args=default_args, catchup=False)
def taskflow():
@task(executor_config=new_config)
def get_testing_increase():
"""
Gets totalTestResultsIncrease field from Covid API for given state and returns value
"""
url = 'https://covidtracking.com/api/v1/states/'
res = requests.get(url+'{0}/current.json'.format(os.environ['STATE']))
return{'testing_increase': json.loads(res.text)['totalTestResultsIncrease']}
get_testing_increase()
dag = taskflow()С помощью new_config мы передаем метку и переменную среды, которую хотим добавить в pod с помощью Kubernetes API. Затем группа DAG выполняет простую задачу, которая использует эту переменную среды для вызова API для получения данных Covid для определенного состояния. Это очень простой пример, но он дает представление о гибкости новой функциональности podOverride. Другие примеры вы можете найти в документации исполнителя Airflow Kubernetes.
Переход на новый KubernetesExecutor
Чтобы упростить внедрение нового KubernetesExecutor, мы предлагаем два варианта адаптации. Первый опыт адаптации предназначен для новых пользователей, а второй — для существующих пользователей.
Для новых пользователей мы предлагаем несколько начальных файлов YAML. Эти файлы предоставляют шаблоны для работы с встроенными DAG, с получением DAG через git и с получением DAG через систему Kubernetes Volume.
Для существующих пользователей мы предлагаем сценарий миграции, который читает ваш airflow.cfg и генерирует файл YAML на основе ваших текущих настроек. Затем этот файл YAML можно изменить для точного соответствия потребностям каждого пользователя.
Лучшая часть этих трех новых функций заключается в том, что все они доступны в Airflow 1.10.13. Вы можете сразу начать процесс миграции и пользоваться преимуществами и ускорением этого более простого дизайна. Мы с нетерпением ждем ваших отзывов и, пожалуйста, не стесняйтесь обращаться с любыми вопросами, запросами функций или документацией!
