Apache airflow запускает операторов, использующих kubernetes в AWS

Question

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

Я обнаружил следующееоператор: KubernetesPodOperator, который, кажется, выполняет эту работу, но единственные примеры, которые я нашел, были в Google Cloud.Я хотел бы запустить это на AWS, однако я не нашел примеров того, как это будет сделано.Я полагаю, что AWS EKS или AWS fargate могут соответствовать требованиям, но не уверены.

Может ли кто-нибудь с опытом воздушного потока, пожалуйста, дайте мне знать, если это возможно?Я посмотрел онлайн и пока не нашел ничего ясного.

Answer 1

Мы использовали Fargate и Airflow в производстве, и опыт пока был хорошим.

Мы использовали его для кратковременных рабочих нагрузок, и это оказалось для нас дешевле, чем выделенный кластер Kubernetes.Кроме того, нет никаких накладных расходов на управление.

Github - DAG с воздушным потоком с ECSOperatorConfig

Answer 2

Вы можете использовать операторы Apache Airflow DAG в любом облачном провайдере, не только в GKE.

Airflow-on-kubernetes-part-1-a-Different-kind-of-operator как Статьи об Airflow Kubernetes Operator предоставляют базовые примеры использования DAG.

Также Исследуйте Airflow KubernetesExecutor на AWS, а статья kops дает хорошее объяснение, с примером накак использовать тома airflow-dags и airflow-logs в AWS.

Пример:

from airflow.operators.python_operator import PythonOperator
from airflow.models import DAG
from datetime import datetime
import time
import os

args = {
    'owner': 'airflow',
    "start_date": datetime(2018, 10, 4),
}

dag = DAG(
    dag_id='test_kubernetes_executor',
    default_args=args,
    schedule_interval=None
)

def print_stuff():
    print("Hi Airflow")

for i in range(2):
    one_task = PythonOperator(
        task_id="one_task" + str(i),
        python_callable=print_stuff,
        dag=dag
    )

    second_task = PythonOperator(
        task_id="two_task" + str(i),
        python_callable=print_stuff,
        dag=dag
    )

    third_task = PythonOperator(
        task_id="third_task" + str(i),
        python_callable=print_stuff,
        dag=dag
    )

    one_task >> second_task >> third_task