Связаться с намиПодключиться

Автоматизация работы с Yandex Data Processing с помощью Yandex Managed Service for Apache Airflow™

Статья создана
Обновлена 16 сентября 2024 г.

В сервисе Yandex Managed Service for Apache Airflow™ можно создать DAG — направленный ациклический граф задач, который позволит автоматизировать работу с сервисом Yandex Data Processing. Ниже рассматривается DAG, который включает в себя несколько задач:

  1. Создать кластер Yandex Data Processing.
  2. Создать и запустить задание PySpark.
  3. Удалить кластер Yandex Data Processing.

При таком DAG кластер существует непродолжительное время. Так как стоимость ресурсов Yandex Data Processing зависит от времени их использования, в кластере можно задействовать ресурсы повышенной мощности и быстро обработать большее количество данных за те же деньги.

В этом DAG кластер Yandex Data Processing создается без сервиса Hive. Для хранения табличных метаданных в примере ниже используется кластер Hive Metastore. Сохраненные метаданные затем может использовать другой кластер Yandex Data Processing.

Чтобы автоматизировать работу с Yandex Data Processing с помощью Managed Service for Apache Airflow™:

  1. Подготовьте инфраструктуру.
  2. Подготовьте PySpark-задание.
  3. Создайте подключения Apache Airflow™.
  4. Подготовьте и запустите DAG-файл.
  5. Проверьте результат.

Если созданные ресурсы вам больше не нужны, удалите их.

Примечание

Сервис Metastore находится на стадии Preview. Чтобы получить доступ, обратитесь в техническую поддержку или к вашему аккаунт-менеджеру.

Подготовьте инфраструктуру

  1. Создайте сервисный аккаунт airflow-sa с ролями:

    • storage.editor;
    • dataproc.agent;
    • dataproc.editor.

  2. Создайте ключи для сервисного аккаунта:

  3. Создайте бакеты:

    • airflow-bucket — для Managed Service for Apache Airflow™;
    • pyspark-bucket — для PySpark-задания;
    • output-bucket – для выходных данных;
    • log-bucket — для сбора логов.

    Бакетов нужно несколько, так как на них назначаются различные права доступа.

  4. Предоставьте сервисному аккаунту airflow-sa разрешения на бакеты:

    • airflow-bucket — разрешение READ;
    • pyspark-bucket — разрешение READ;
    • output-bucket – разрешение READ и WRITE;
    • log-bucket — разрешение READ и WRITE.

  5. Создайте облачную сеть с именем dataproc-network.

    Вместе с ней автоматически создадутся три подсети в разных зонах доступности и группа безопасности.

  6. Настройте NAT-шлюз для подсети dataproc-network-ru-central1-a.

  7. Добавьте правила в группу безопасности в сети dataproc-network:

    Правила группы безопасности

    Для какого сервиса нужно правило

    Зачем нужно правило

    Настройки правила

    Yandex Data Processing

    Для входящего служебного трафика.
    • Диапазон портов — 0-65535.
    • Протокол — Любой (Any).
    • Источник — Группа безопасности.
    • Группа безопасности — Текущая (Self).

    Metastore

    Для входящего трафика от клиентов.
    • Диапазон портов — 30000-32767.
    • Протокол — Любой (Any).
    • Источник — CIDR.
    • CIDR блоки — 0.0.0.0/0.

    Metastore

    Для входящего трафика от балансировщика.
    • Диапазон портов — 10256.
    • Протокол — Любой (Any).
    • Источник — Проверки состояния балансировщика.

    Yandex Data Processing

    Для исходящего служебного трафика.
    • Диапазон портов — 0-65535.
    • Протокол — Любой (Any).
    • Источник — Группа безопасности.
    • Группа безопасности — Текущая (Self).

    Yandex Data Processing

    Для исходящего HTTPS-трафика.
    • Диапазон портов — 443.
    • Протокол — TCP.
    • Назначение — CIDR.
    • CIDR блоки — 0.0.0.0/0.

    Yandex Data Processing

    Для исходящего трафика, чтобы разрешить подключение кластера Yandex Data Processing к Metastore.
    • Диапазон портов — 9083.
    • Протокол — Любой (Any).
    • Источник — CIDR.
    • CIDR блоки — 0.0.0.0/0.

  8. Создайте кластер Metastore с параметрами:

    • Сеть — dataproc-network.
    • Подсеть — dataproc-network-ru-central1-a.
    • Группа безопасности — группа по умолчанию в сети dataproc-network.
    • Идентификатор ключа и секретный ключ — принадлежат статическому ключу доступа.

  9. Создайте кластер Managed Service for Apache Airflow™ с параметрами:

    • Зона доступности — ru-central1-a.
    • Сеть — dataproc-network.
    • Подсеть — dataproc-network-ru-central1-a.
    • Группа безопасности — группа по умолчанию в сети dataproc-network.
    • Имя бакета — airflow-bucket.

Подготовьте PySpark-задание

Для PySpark-задания будет использован Python-скрипт, который создает таблицу и хранится в бакете Object Storage. Подготовьте файл скрипта:

  1. Создайте локально файл с именем create-table.py и скопируйте в него скрипт:

    create-table.py
    from pyspark.sql.types import *
from pyspark.sql import SparkSession

# Создание Spark-сессии
spark = SparkSession.builder \
    .appName("create-table") \
    .enableHiveSupport() \
    .getOrCreate()

# Создание схемы данных
schema = StructType([StructField('Name', StringType(), True),
StructField('Capital', StringType(), True),
StructField('Area', IntegerType(), True),
StructField('Population', IntegerType(), True)])

# Создание датафрейма
df = spark.createDataFrame([('Австралия', 'Канберра', 7686850, 19731984), ('Австрия', 'Вена', 83855, 7700000)], schema)

# Запись датафрейма в бакет в виде таблицы countries
df.write.mode("overwrite").option("path","s3a://output-bucket/countries").saveAsTable("countries")

  2. Создайте в бакете pyspark-bucket папку scripts и загрузите в нее файл create-table.py.

Создайте подключения Apache Airflow™

Для DAG задач будет использовано два подключения: для Object Storage и сервисного аккаунта. Они будут указаны в DAG-файле. Чтобы создать подключения:

  1. Откройте веб-интерфейс Apache Airflow™.

  2. Перейдите в раздел Admin → Connections.

  3. Нажмите кнопку image.

  4. Создайте подключение для Object Storage. Укажите его параметры:

    • Connection Id — yc-s3.

    • Connection Type — Amazon Elastic MapReduce.

    • Run Job Flow Configuration — JSON-файл в формате:

      {
   "aws_access_key_id": "<идентификатор_статического_ключа>",
   "aws_secret_access_key": "<секретный_ключ>",
   "host": "https://storage.yandexcloud.net/"
}

      В файле укажите данные статического ключа, который вы создавали ранее.

  5. Нажмите кнопку Save.

  6. Снова нажмите кнопку image, чтобы создать подключение для сервисного аккаунта.

  7. Укажите параметры подключения:

    • Connection Id — yc-airflow-sa.
    • Connection Type — Yandex Cloud.
    • Service account auth JSON — содержимое JSON-файла с авторизованным ключом.
    • Public SSH key — публичная часть авторизованного ключа, который вы создавали ранее. Оформите ключ в виде одной строки, в нем не должно быть символов переноса.

    Остальные поля оставьте пустыми.

  8. Нажмите кнопку Save.

Подготовьте и запустите DAG-файл

DAG будет состоять из нескольких вершин, которые формируют цепочку последовательных действий:

  1. Managed Service for Apache Airflow™ создает временный, легковесный кластер Yandex Data Processing с настройками, заданными в DAG. Этот кластер автоматически подключается к созданному ранее кластеру Metastore.
  2. Когда кластер Yandex Data Processing готов, запускается задание PySpark.
  3. После выполнения задания временный кластер Yandex Data Processing удаляется.

Чтобы подготовить DAG:

  1. Создайте SSH-ключ. Сохраните открытую часть ключа — она понадобится для создания кластера Yandex Data Processing.

  2. Создайте локально файл с именем Data-Proc-DAG.py, скопируйте в него скрипт и подставьте данные вашей инфраструктуры в переменные:

    Data-Proc-DAG.py
    import uuid
import datetime
from airflow import DAG, settings
from airflow.models import Connection, Variable
from airflow.utils.trigger_rule import TriggerRule
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.providers.amazon.aws.operators.s3 import S3DeleteObjectsOperator
from airflow.providers.yandex.operators.yandexcloud_dataproc import (
    DataprocCreateClusterOperator,
    DataprocCreatePysparkJobOperator,
    DataprocDeleteClusterOperator,
)

# Данные вашей инфраструктуры
YC_DP_FOLDER_ID = '<идентификатор_каталога>'
YC_DP_SSH_PUBLIC_KEY = '<открытая_часть_SSH-ключа>'
YC_DP_SUBNET_ID = '<идентификатор_подсети>'
YC_DP_GROUP_ID = '<идентификатор_группы_безопасности>'
YC_DP_SA_ID = '<идентификатор_сервисного_аккаунта>'
YC_DP_METASTORE_URI = '<IP-адрес>'
YC_DP_AZ = 'ru-central1-a'
YC_SOURCE_BUCKET = 'pyspark-bucket'
YC_DP_LOGS_BUCKET = 'log-bucket'

# Создание подключения для Object Storage
session = settings.Session()
ycS3_connection = Connection(
    conn_id='yc-s3'
)
if not session.query(Connection).filter(Connection.conn_id == ycS3_connection.conn_id).first():
    session.add(ycS3_connection)
    session.commit()

# Создание подключения для сервисного аккаунта
ycSA_connection = Connection(
    conn_id='yc-airflow-sa'
)
if not session.query(Connection).filter(Connection.conn_id == ycSA_connection.conn_id).first():
    session.add(ycSA_connection)
    session.commit()

# Настройки DAG
with DAG(
        'DATA_INGEST',
        schedule_interval='@hourly',
        tags=['data-proc-and-airflow'],
        start_date=datetime.datetime.now(),
        max_active_runs=1,
        catchup=False
) as ingest_dag:

    # 1 этап: создание кластера Yandex Data Processing
    create_spark_cluster = DataprocCreateClusterOperator(
        task_id='dp-cluster-create-task',
        folder_id=YC_DP_FOLDER_ID,
        cluster_name=f'tmp-dp-{uuid.uuid4()}',
        cluster_description='Временный кластер для выполнения PySpark-задания под оркестрацией Managed Service for Apache Airflow™',
        ssh_public_keys=YC_DP_SSH_PUBLIC_KEY,
        subnet_id=YC_DP_SUBNET_ID,
        s3_bucket=YC_DP_LOGS_BUCKET,
        service_account_id=YC_DP_SA_ID,
        zone=YC_DP_AZ,
        cluster_image_version='2.1.7',
        enable_ui_proxy=False,
        masternode_resource_preset='s2.small',
        masternode_disk_type='network-ssd',
        masternode_disk_size=200,
        computenode_resource_preset='m2.large',
        computenode_disk_type='network-ssd',
        computenode_disk_size=200,
        computenode_count=2,
        computenode_max_hosts_count=5,  # Количество подкластеров для обработки данных будет автоматически масштабироваться в случае большой нагрузки.
        services=['YARN', 'SPARK'],     # Создается легковесный кластер.
        datanode_count=0,               # Без подкластеров для хранения данных.
        properties={                    # С указанием на удаленный кластер Metastore.
            'spark:spark.hive.metastore.uris': f'thrift://{YC_DP_METASTORE_URI}:9083',
        },
        security_group_ids=[YC_DP_GROUP_ID],
        connection_id=ycSA_connection.conn_id,
        dag=ingest_dag
    )

    # 2 этап: запуск задания PySpark
    poke_spark_processing = DataprocCreatePysparkJobOperator(
        task_id='dp-cluster-pyspark-task',
        main_python_file_uri=f's3a://{YC_SOURCE_BUCKET}/scripts/create-table.py',
        connection_id=ycSA_connection.conn_id,
        dag=ingest_dag
    )

    # 3 этап: удаление кластера Yandex Data Processing
    delete_spark_cluster = DataprocDeleteClusterOperator(
        task_id='dp-cluster-delete-task',
        trigger_rule=TriggerRule.ALL_DONE,
        dag=ingest_dag
    )

    # Формирование DAG из указанных выше этапов
    create_spark_cluster >> poke_spark_processing >> delete_spark_cluster

    Где:

    • YC_DP_FOLDER_ID — идентификатор каталога в облаке.
    • YC_DP_SSH_PUBLIC_KEY — открытая часть SSH-ключа для кластера Yandex Data Processing.
    • YC_DP_SUBNET_ID — идентификатор подсети.
    • YC_DP_GROUP_ID — идентификатор группы безопасности.
    • YC_DP_SA_ID — идентификатор сервисного аккаунта.
    • YC_DP_METASTORE_URI — IP-адрес кластера Metastore.
    • YC_DP_AZ — зона доступности для кластера Yandex Data Processing, например ru-central1-a.
    • YC_SOURCE_BUCKET — бакет с Python-скриптом для задания PySpark, например pyspark-bucket.
    • YC_DP_LOGS_BUCKET — бакет для логов, например pyspark-bucket.

  3. Загрузите DAG в кластер Managed Service for Apache Airflow™: создайте в бакете airflow-bucket папку dags и загрузите в нее файл Data-Proc-DAG.py.

  4. Откройте веб-интерфейс Apache Airflow™.

  5. Убедитесь, что в разделе DAGs появился новый DAG DATA_INGEST с тегом data-proc-and-airflow.

    Загрузка DAG-файла из бакета может занять несколько минут.

  6. Чтобы запустить DAG, в строке с его именем нажмите кнопку image.

Проверьте результат

  1. Чтобы отслеживать результаты выполнения задач, нажмите на название DAG. Результаты отображаются во вкладке Grid.
  2. Дождитесь, когда все три задачи в DAG перейдут в статус Success. Параллельно вы можете проверить, что в консоли управления создается кластер Yandex Data Processing, выполняется задание PySpark и удаляется тот же кластер.
  3. Убедитесь, что в бакете output-bucket появилась папка countries, а в ней — файл part-00000-.... Теперь данные из созданной таблицы хранятся в бакете Object Storage, а метаинформация о ней — в кластере Metastore.
  4. Проверьте, что в бакете log-bucket появились логи выполнения PySpark-задания.

Удалите созданные ресурсы

Некоторые ресурсы платные. Удалите ресурсы, которые вы больше не будете использовать, чтобы не платить за них:

  1. Сервисный аккаунт.
  2. Бакеты Object Storage.
  3. Кластер Metastore.
  4. Кластер Managed Service for Apache Airflow™.
  5. Таблицу маршрутизации.
  6. NAT-шлюз.
  7. Группу безопасности.
  8. Облачные подсети, созданные по умолчанию в сети dataproc-network.
  9. Облачную сеть.
Предыдущая
Работа с топиками Apache Kafka® с помощью Yandex Data Processing
Следующая
Совместная работа с таблицами Yandex Data Processing с использованием Metastore