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

Обмен данными между Yandex Managed Service for ClickHouse® и Yandex Data Processing

Статья создана
Обновлена 8 апреля 2025 г.

С помощью Yandex Data Processing вы можете:

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

В стоимость поддержки описываемого решения входят:

  • Плата за кластер Yandex Data Processing: использование вычислительных ресурсов ВМ и сетевых дисков Compute Cloud, а также сервиса Cloud Logging для работы с логами (см. тарифы Yandex Data Processing).
  • Плата за кластер Managed Service for ClickHouse®: использование вычислительных ресурсов, выделенных хостам (в том числе хостам ZooKeeper), и дискового пространства (см. тарифы Managed Service for ClickHouse®).
  • Плата за NAT-шлюз (см. тарифы Virtual Private Cloud).
  • Плата за бакет Object Storage: хранение данных и выполнение операций с ними (см. тарифы Object Storage).
  • Плата за публичные IP-адреса для хостов кластеров (см. тарифы Virtual Private Cloud).

Перед началом работы

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

  1. Создайте сервисный аккаунт с именем dataproc-sa и назначьте ему роли dataproc.agent и dataproc.provisioner.

  2. В Object Storage создайте бакеты и настройте доступ к ним:

    1. Создайте бакет для исходных данных и предоставьте сервисному аккаунту кластера разрешение READ для этого бакета.
    2. Создайте бакет для результатов обработки и предоставьте сервисному аккаунту кластера разрешение READ и WRITE для этого бакета.

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

  4. В сети dataproc-network создайте подсеть в любой зоне доступности.

  5. Настройте NAT-шлюз для созданной подсети.

  6. Если вы используете группы безопасности, создайте группу безопасности с именем dataproc-sg в сети dataproc-network и добавьте в нее следующие правила:

    • По одному правилу для входящего и исходящего служебного трафика:

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

    • Правило для исходящего HTTPS-трафика:

      • Диапазон портов — 443.
      • Протокол — TCP.
      • Назначение — CIDR.
      • CIDR блоки — 0.0.0.0/0.

    • Правило для исходящего трафика по протоколу TCP на порт 8443 для доступа к ClickHouse®:

      • Диапазон портов — 8443.
      • Протокол — TCP.
      • Назначение — CIDR.
      • CIDR блоки — 0.0.0.0/0.

  7. Создайте кластер Yandex Data Processing с любой подходящей конфигурацией хостов и следующими настройками:

    • Компоненты:
      • SPARK;
      • YARN;
      • HDFS.
    • Сервисный аккаунт — dataproc-sa.
    • Имя бакета — бакет, который вы создали для выходных данных.
    • Сеть — dataproc-network.
    • Группы безопасности — dataproc-sg.

  8. Создайте кластер Managed Service for ClickHouse® любой подходящей конфигурации со следующими настройками:

    • С публичным доступом к хостам кластера.
    • С базой данных db1.
    • С пользователем user1.

  9. Если вы используете группы безопасности в кластере Managed Service for ClickHouse®, убедитесь, что они настроены правильно и допускают подключение к нему.

  1. Если у вас еще нет Terraform, установите его.

  2. Получите данные для аутентификации. Вы можете добавить их в переменные окружения или указать далее в файле с настройками провайдера.

  3. Настройте и инициализируйте провайдер. Чтобы не создавать конфигурационный файл с настройками провайдера вручную, скачайте его.

  4. Поместите конфигурационный файл в отдельную рабочую директорию и укажите значения параметров. Если данные для аутентификации не были добавлены в переменные окружения, укажите их в конфигурационном файле.

  5. Скачайте в ту же рабочую директорию файл конфигурации data-proc-data-exchange-with-mch.tf.

    В этом файле описаны:

    • сеть;
    • подсеть;
    • NAT-шлюз и таблица маршрутизации, необходимые для работы Yandex Data Processing;
    • группы безопасности, необходимые для кластеров Yandex Data Processing и Managed Service for ClickHouse®;
    • сервисный аккаунт, необходимый для работы кластера Yandex Data Processing;
    • сервисный аккаунт, необходимый для создания бакетов в Object Storage;
    • бакеты для входных и выходных данных;
    • кластер Yandex Data Processing;
    • кластер Managed Service for ClickHouse®.

  6. Укажите в файле data-proc-data-exchange-with-mch.tf:

    • folder_id — идентификатор облачного каталога, такой же как в настройках провайдера.
    • input_bucket — имя бакета для входных данных.
    • output_bucket — имя бакета для выходных данных.
    • dp_ssh_key — абсолютный путь к публичному ключу для кластера Yandex Data Processing. Подробнее см. в разделе SSH-подключение к хосту Yandex Data Processing.
    • ch_password — пароль пользователя ClickHouse®.

  7. Проверьте корректность файлов конфигурации Terraform с помощью команды:

    terraform validate

    Если в файлах конфигурации есть ошибки, Terraform на них укажет.

  8. Создайте необходимую инфраструктуру:

    1. Выполните команду для просмотра планируемых изменений:

      terraform plan

      Если конфигурации ресурсов описаны верно, в терминале отобразится список изменяемых ресурсов и их параметров. Это проверочный этап: ресурсы не будут изменены.

    2. Если вас устраивают планируемые изменения, внесите их:

      1. Выполните команду:

        terraform apply

      2. Подтвердите изменение ресурсов.

      3. Дождитесь завершения операции.

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

Загрузите данные из Managed Service for ClickHouse®

Подготовьте таблицу в кластере Managed Service for ClickHouse®

  1. Подключитесь к базе данных db1 кластера Managed Service for ClickHouse® от имени пользователя user1.

  2. Наполните базу тестовыми данными. В качестве примера используется простая таблица с именами и возрастом людей.

    1. Создайте таблицу:

      CREATE TABLE persons (
    `name` String,
    `age` UInt8) ENGINE = MergeTree ()
ORDER BY
    `name`;

    2. Наполните таблицу данными:

      INSERT INTO persons VALUES
   ('Anna', 19),
   ('Michael', 65),
   ('Alvar', 28),
   ('Lilith', 50),
   ('Max', 27),
   ('Jaimey', 34),
   ('Dmitry', 42),
   ('Qiang', 19),
   ('Augustyna', 20),
   ('Maria', 28);

    3. Проверьте результат:

      SELECT * FROM persons;

Перенесите таблицу из Managed Service for ClickHouse®

  1. Подготовьте файл скрипта:

    1. Создайте локальный файл с именем ch-to-dataproc.py и скопируйте в него следующий скрипт:

      ch-to-dataproc.py
      from pyspark.sql import SparkSession

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

# Указание порта и параметров кластера ClickHouse®
jdbcPort = 8443
jdbcHostname = "c-<идентификатор_кластера_ClickHouse®>.rw.mdb.yandexcloud.net"
jdbcDatabase = "db1"
jdbcUrl = f"jdbc:clickhouse://{jdbcHostname}:{jdbcPort}/{jdbcDatabase}?ssl=true"

# Перенос таблицы persons из ClickHouse® в DataFrame
df = spark.read.format("jdbc") \
.option("url", jdbcUrl) \
.option("user","user1") \
.option("password","<пароль_пользователя_user1>") \
.option("dbtable","persons") \
.load()

# Перенос DataFrame в бакет для проверки
df.repartition(1).write.mode("overwrite") \
.csv(path='s3a://<имя_выходного_бакета>/csv', header=True, sep=',')

    2. Укажите в скрипте:

      • Идентификатор кластера Managed Service for ClickHouse®.
      • Пароль пользователя user1.
      • Имя выходного бакета.

    3. Создайте в бакете для входных данных папку scripts и загрузите в нее файл ch-to-dataproc.py.

  2. Создайте задание PySpark, указав в поле Main python файл путь к файлу скрипта: s3a://<имя_входного_бакета>/scripts/ch-to-dataproc.py.

  3. Дождитесь завершения задания и проверьте, что в папке csv выходного бакета появилась исходная таблица.

Примечание

Вы можете просматривать логи выполнения заданий и искать в них информацию с помощью сервиса Yandex Cloud Logging. Подробнее см. в разделе Работа с логами.

Выгрузите данные в Managed Service for ClickHouse®

  1. Подготовьте файл скрипта:

    1. Создайте локальный файл с именем dataproc-to-ch.py и скопируйте в него следующий скрипт:

      dataproc-to-ch.py
      from pyspark.sql import SparkSession
from pyspark.sql.types import *

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

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

# Создание DataFrame
df = spark.createDataFrame([('Alim', 19),
                            ('Fred' ,65),
                            ('Guanmin' , 28),
                            ('Till', 60),
                            ('Almagul', 27),
                            ('Mary', 34),
                            ('Dmitry', 42)], schema)

# Указание порта и параметров кластера ClickHouse®
jdbcPort = 8443
jdbcHostname = "c-<идентификатор_кластера_ClickHouse®>.rw.mdb.yandexcloud.net"
jdbcDatabase = "db1"
jdbcUrl = f"jdbc:clickhouse://{jdbcHostname}:{jdbcPort}/{jdbcDatabase}?ssl=true"

# Перенос DataFrame в ClickHouse®
df.write.format("jdbc") \
.mode("error") \
.option("url", jdbcUrl) \
.option("dbtable", "people") \
.option("createTableOptions", "ENGINE = MergeTree() ORDER BY age") \
.option("user","user1") \
.option("password","<пароль_к_базе_данных_ClickHouse®>") \
.save()

    2. Укажите в скрипте:

      • Идентификатор кластера Managed Service for ClickHouse®.
      • Пароль пользователя user1.

    3. Создайте в бакете для входных данных папку scripts и загрузите в нее файл dataproc-to-ch.py.

  2. Создайте задание PySpark, указав в поле Main python файл путь к файлу скрипта: s3a://<имя_входного_бакета>/scripts/dataproc-to-ch.py.

  3. Дождитесь завершения задания и проверьте, что данные перенеслись в Managed Service for ClickHouse®:

    1. Подключитесь к базе данных db1 кластера Managed Service for ClickHouse® от имени пользователя user1.

    2. Выполните запрос:

      SELECT * FROM people;

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

Примечание

Вы можете просматривать логи выполнения заданий и искать в них информацию с помощью сервиса Yandex Cloud Logging. Подробнее см. в разделе Работа с логами.

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

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

  1. Удалите объекты из бакетов. Остальные ресурсы удалите в зависимости от способа их создания:

    1. В терминале перейдите в директорию с планом инфраструктуры.

      Важно

      Убедитесь, что в директории нет Terraform-манифестов с ресурсами, которые вы хотите сохранить. Terraform удаляет все ресурсы, которые были созданы с помощью манифестов в текущей директории.

    2. Удалите ресурсы:

      1. Выполните команду:

        terraform destroy

      2. Подтвердите удаление ресурсов и дождитесь завершения операции.

      Все ресурсы, которые были описаны в Terraform-манифестах, будут удалены.

ClickHouse® является зарегистрированным товарным знаком ClickHouse, Inc.

Предыдущая
Загрузка данных в Yandex Managed Service for YDB с помощью Yandex Data Transfer
Следующая
Загрузка данных из Yandex Managed Service for YDB с помощью Yandex Data Transfer