Интеграция Amazon Glue с ClickHouse и Spark

ClickHouse Supported

Amazon Glue — это полностью управляемый бессерверный сервис для интеграции данных от Amazon Web Services (AWS). Он упрощает процесс обнаружения, подготовки и преобразования данных для аналитики, машинного обучения и разработки приложений.

Установка

Чтобы интегрировать ваш код Glue с ClickHouse, вы можете использовать наш официальный коннектор Spark в Glue одним из следующих способов:

• Установить коннектор ClickHouse Glue из AWS Marketplace (рекомендуется).
• Вручную добавить JAR‑файлы Spark Connector в ваше задание Glue.

AWS Marketplace

1. Подпишитесь на коннектор

   Чтобы получить доступ к коннектору в вашей учётной записи, оформите подписку на ClickHouse AWS Glue Connector в AWS Marketplace.

2. Предоставьте необходимые разрешения

   Убедитесь, что роль IAM вашего задания Glue имеет необходимые разрешения, как описано в руководстве по минимальным привилегиям здесь.

3. Активируйте коннектор и создайте подключение

   Вы можете активировать коннектор и создать подключение напрямую, нажав эту ссылку, которая откроет страницу создания подключения Glue с заранее заполненными ключевыми полями. Задайте подключению имя и нажмите кнопку Create (на этом этапе нет необходимости указывать параметры подключения к ClickHouse).

4. Использование в задании Glue

   В вашем задании Glue выберите вкладку Job details и разверните окно Advanced properties. В разделе Connections выберите только что созданное подключение. Коннектор автоматически добавит необходимые JAR‑файлы в среду выполнения задания.

Примечание

JAR‑файлы, используемые в коннекторе Glue, собраны для Spark 3.3, Scala 2 и Python 3. Убедитесь, что вы выбираете эти версии при настройке вашего задания Glue.

Ручная установка

Чтобы добавить необходимые JAR‑файлы вручную, выполните следующее:

1. Загрузите следующие JAR‑файлы в бакет S3: clickhouse-jdbc-0.6.X-all.jar и clickhouse-spark-runtime-3.X_2.X-0.8.X.jar.
2. Убедитесь, что задание Glue имеет доступ к этому бакету.
3. На вкладке Job details пролистайте вниз, разверните выпадающий список Advanced properties и укажите путь к JAR‑файлам в поле Dependent JARs path:

Примеры

Scala

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.GlueArgParser
import com.amazonaws.services.glue.util.Job
import com.clickhouseScala.Native.NativeSparkRead.spark
import org.apache.spark.sql.SparkSession

import scala.collection.JavaConverters._
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

object ClickHouseGlueExample {
  def main(sysArgs: Array[String]) {
    val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray)

    val sparkSession: SparkSession = SparkSession.builder
      .config("spark.sql.catalog.clickhouse", "com.clickhouse.spark.ClickHouseCatalog")
      .config("spark.sql.catalog.clickhouse.host", "<your-clickhouse-host>")
      .config("spark.sql.catalog.clickhouse.protocol", "https")
      .config("spark.sql.catalog.clickhouse.http_port", "<your-clickhouse-port>")
      .config("spark.sql.catalog.clickhouse.user", "default")
      .config("spark.sql.catalog.clickhouse.password", "<your-password>")
      .config("spark.sql.catalog.clickhouse.database", "default")
      // для ClickHouse Cloud
      .config("spark.sql.catalog.clickhouse.option.ssl", "true")
      .config("spark.sql.catalog.clickhouse.option.ssl_mode", "NONE")
      .getOrCreate

    val glueContext = new GlueContext(sparkSession.sparkContext)
    Job.init(args("JOB_NAME"), glueContext, args.asJava)
    import sparkSession.implicits._

    val url = "s3://{path_to_cell_tower_data}/cell_towers.csv.gz"

    val schema = StructType(Seq(
      StructField("radio", StringType, nullable = false),
      StructField("mcc", IntegerType, nullable = false),
      StructField("net", IntegerType, nullable = false),
      StructField("area", IntegerType, nullable = false),
      StructField("cell", LongType, nullable = false),
      StructField("unit", IntegerType, nullable = false),
      StructField("lon", DoubleType, nullable = false),
      StructField("lat", DoubleType, nullable = false),
      StructField("range", IntegerType, nullable = false),
      StructField("samples", IntegerType, nullable = false),
      StructField("changeable", IntegerType, nullable = false),
      StructField("created", TimestampType, nullable = false),
      StructField("updated", TimestampType, nullable = false),
      StructField("averageSignal", IntegerType, nullable = false)
    ))

    val df = sparkSession.read
      .option("header", "true")
      .schema(schema)
      .csv(url)

    // Запись в ClickHouse
    df.writeTo("clickhouse.default.cell_towers").append()


    // Чтение из ClickHouse
    val dfRead = spark.sql("select * from clickhouse.default.cell_towers")
    Job.commit()
  }
}

Python

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql import Row


## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
logger = glueContext.get_logger()
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

spark.conf.set("spark.sql.catalog.clickhouse", "com.clickhouse.spark.ClickHouseCatalog")
spark.conf.set("spark.sql.catalog.clickhouse.host", "<your-clickhouse-host>")
spark.conf.set("spark.sql.catalog.clickhouse.protocol", "https")
spark.conf.set("spark.sql.catalog.clickhouse.http_port", "<your-clickhouse-port>")
spark.conf.set("spark.sql.catalog.clickhouse.user", "default")
spark.conf.set("spark.sql.catalog.clickhouse.password", "<your-password>")
spark.conf.set("spark.sql.catalog.clickhouse.database", "default")
spark.conf.set("spark.clickhouse.write.format", "json")
spark.conf.set("spark.clickhouse.read.format", "arrow")
# для ClickHouse Cloud
spark.conf.set("spark.sql.catalog.clickhouse.option.ssl", "true")
spark.conf.set("spark.sql.catalog.clickhouse.option.ssl_mode", "NONE")

# Создать DataFrame
data = [Row(id=11, name="John"), Row(id=12, name="Doe")]
df = spark.createDataFrame(data)

# Записать DataFrame в ClickHouse
df.writeTo("clickhouse.default.example_table").append()

# Прочитать DataFrame из ClickHouse
df_read = spark.sql("select * from clickhouse.default.example_table")
logger.info(str(df.take(10)))

job.commit()

Подробности см. в нашей документации по Spark.