Подключиться
Практические руководства

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

Статья создана

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

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

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

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

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

  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.

    Функциональность находится на стадии Preview.

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

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

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

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

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

    Функциональность находится на стадии Preview.

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

  2. Скачайте файл с настройками провайдера. Поместите его в отдельную рабочую директорию и укажите значения параметров.

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

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

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

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

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

  5. Выполните команду terraform init в директории с конфигурационными файлами. Эта команда инициализирует провайдер, указанный в конфигурационном файле, и позволяет работать с ресурсами и источниками данных провайдера.

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

    terraform validate

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

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

    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-<идентификатор кластера Managed Service for 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 выходного бакета появилась исходная таблица.

Выгрузите данные в 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-<идентификатор кластера Managed Service for 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","<пароль к базе данных Managed Service for 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;

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

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

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

  1. Удалите объекты из бакетов.

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

  3. Удалите конфигурационный файл data-proc-data-exchange-with-mch.tf.

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

    terraform validate

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

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

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

      terraform plan

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

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

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

        terraform apply

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

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

    Все ресурсы, которые были описаны в конфигурационном файле data-proc-data-exchange-with-mch.tf, будут удалены.

В этой статье: