Какие операторы Apache Airflow вы использовали?

В своих проектах я активно использовал как стандартные, так и специфичные для облака операторы Airflow для оркестрации разнообразных задач:

• BashOperator & PythonOperator: Для запуска shell-скриптов и вызова Python-функций соответственно. Это основа многих DAG.

  from airflow.operators.bash import BashOperator
  from airflow.operators.python import PythonOperator
  
  def _process_data(ti, **kwargs):
      # Получаем результат предыдущей задачи через XCom
      raw_data = ti.xcom_pull(task_ids='extract_data')
      processed_data = transform_logic(raw_data)
      return processed_data
  
  process_task = PythonOperator(
      task_id='process_data',
      python_callable=_process_data,
      provide_context=True
  )
  
  cleanup_task = BashOperator(
      task_id='cleanup_temp_files',
      bash_command='rm -f /tmp/*.tmp'
  )

• Сенсоры (Sensors): Использовал S3KeySensor для ожидания появления файла-триггера в S3 и ExternalTaskSensor для ожидания завершения другого DAG.

• Операторы для работы с БД: PostgresOperator для выполнения SQL в PostgreSQL и BigQueryExecuteQueryOperator для Google BigQuery.

  from airflow.providers.postgres.operators.postgres import PostgresOperator
  
  create_table = PostgresOperator(
      task_id='create_staging_table',
      postgres_conn_id='analytics_db',
      sql="""
      CREATE TABLE IF NOT EXISTS stg_orders (
          id INT,
          created_at TIMESTAMP,
          amount DECIMAL
      );
      """
  )

• Операторы для контейнеризованных и распределённых задач:

  • DockerOperator для запуска задач в изолированных контейнерах с определённым образом.
  • SparkSubmitOperator для отправки Spark-приложений на кластер YARN или Kubernetes. Это было ключевым для ETL-пайплайнов.

• KubernetesPodOperator: Для запуска задач в виде подов в Kubernetes, что обеспечивало гибкость в выборе окружения и ресурсов для каждой задачи.