Обмен данными с Yandex Data Processing

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

• Загрузить данные из Managed Service for ClickHouse® в Spark DataFrame.
• Выгрузить данные из Spark DataFrame в Managed Service for ClickHouse®.

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

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

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

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

Подготовьте таблицу в кластере 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®Перенесите таблицу из 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 выходного бакета появилась исходная таблица.

Выгрузите данные в Managed Service for ClickHouse®Выгрузите данные в 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;
      

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

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

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

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