Импорт данных из Yandex Object Storage, обработка и экспорт в Yandex Managed Service for ClickHouse®

В качестве примера используются две CSV-таблицы, которые нужно объединить в одну, импортировать в формат Parquet и передать в Managed Service for ClickHouse®.

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

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

Подготовьте тестовые данныеПодготовьте тестовые данные

Для примера используются две таблицы в формате CSV:

• coords.csv — содержит информацию о географических координатах автомобиля.
• sensors.csv — содержит информацию о скорости и рабочих параметрах автомобиля.

Чтобы подготовить тестовые данные к работе:

1. Скопируйте содержимое файлов из примеров ниже и сохраните их локально в формате CSV:

   • coords.csv

     vehicle_id,latitude,longitude,altitude
     iv9a94th6rztooxh5ur2,55.70329032,37.65472196,427.5
     022wsiz48h2ljxuz04x8,56.96149325,38.46541766,423.6
     a7fbbqjws4zqw85f6jue,54.99296663,36.79063999,426.2
     l7731117m6r6on4m633n,55.34740545,37.13175678,422.5
     6f9q6ienc4qfpdwd9nef,56.69752218,38.38871530,428.3
     

   • sensors.csv

     vehicle_id,speed,battery_voltage,cabin_temperature,fuel_level
     iv9a94th6rztooxh5ur2,0.0,25.5,17,5
     022wsiz48h2ljxuz04x8,55.5,54.5,21,22
     a7fbbqjws4zqw85f6jue,80.6,22.1,19,73
     l7731117m6r6on4m633n,40.9,76.0,25,23
     6f9q6ienc4qfpdwd9nef,64.8,90.8,21,32
     

2. Создайте в бакете для входных данных папку csv и загрузите в нее созданные CSV-файлы.

Обработайте данные в Yandex Data ProcessingОбработайте данные в Yandex Data Processing

Объедините данные из двух таблиц в одну и загрузите ее в формате Parquet в бакет, который вы ранее создали для результатов обработки:

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

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

      join-tables.py

      from pyspark.sql import SparkSession
      
      # Создание Spark-сессии
      spark = SparkSession.builder.appName("JoinExample").getOrCreate()
      
      # Чтение таблицы из файла coords.csv
      coords_df = spark.read.csv("s3a://<имя_входного_бакета>/csv/coords.csv", header=True)
      
      # Чтение таблицы из файла sensors.csv
      sensors_df = spark.read.csv("s3a://<имя_входного_бакета>/csv/sensors.csv", header=True)
      
      # Объединение таблицы по столбцу vehicle_id
      joined_df = coords_df.join(sensors_df, on="vehicle_id", how="inner")
      
      # Сохранение объединенной таблицы в бакет в формате Parquet
      joined_df.write.parquet("s3a://<имя_выходного_бакета>/parquet/")
      

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

      • Имя входного бакета, в котором хранятся исходные CSV-таблицы.
      • Имя выходного бакета, в который будет сохранен Parquet-файл с объединенными данными.

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

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

3. Дождитесь завершения задания и проверьте, что в выходном бакете в папке parquet появился Parquet-файл part-00000-***.

Экспортируйте данные в ClickHouse®Экспортируйте данные в ClickHouse®

Перенесите объединенную таблицу из Object Storage в ClickHouse®:

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

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

      parquet-to-ch.py

      from pyspark.sql import SparkSession
      
      # Создание Spark-сессии
      spark = SparkSession.builder.appName("ParquetClickhouse").getOrCreate()
      
      # Чтение данных из Parquet-файла
      parquetFile = spark.read.parquet("s3a://<имя_выходного_бакета>/parquet/*.parquet")
      
      # Указание порта и параметров кластера ClickHouse®
      jdbcPort = 8443
      jdbcHostname = "c-<идентификатор_кластера>.rw.mdb.yandexcloud.net"
      jdbcDatabase = "db1"
      jdbcUrl = f"jdbc:clickhouse://{jdbcHostname}:{jdbcPort}/{jdbcDatabase}?ssl=true"
      
      # Перенос таблицы из Parquet-файла в ClickHouse®-таблицу с именем measurements
      parquetFile.write.format("jdbc") \
      .mode("error") \
      .option("url", jdbcUrl) \
      .option("dbtable", "measurements") \
      .option("createTableOptions", "ENGINE = MergeTree() ORDER BY vehicle_id") \
      .option("user","user1") \
      .option("password","<пароль_пользователя_ClickHouse®>") \
      .save()
      

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

      • Имя бакета, в котором лежит Parquet-файл.
      • Идентификатор кластера Managed Service for ClickHouse®.
      • Пароль пользователя ClickHouse®.

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

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

3. Дождитесь выполнения задания и убедитесь, что объединенная таблица перенесена в кластер:

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

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

      SELECT * FROM measurements;
      

   Если экспорт данных прошел успешно, ответом на запрос будет объединенная таблица.

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

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

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