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

Поддерживается в ClickHouse

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

Установка

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

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

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

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

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

После оформления подписки выберите версию Glue, соответствующую требованиям вашей задачи. В разделе Additional details, в подразделе Usage instructions нажмите ссылку Open Glue Studio - Add ClickHouse connector. Откроется страница создания подключения Glue с уже заполненными ключевыми полями. Задайте имя подключения и нажмите Create (на этом этапе указывать сведения о подключении к ClickHouse не нужно).

AWS Marketplace usage instructions for ClickHouse Glue connector

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

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

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

Загрузите JAR-файл коннектора

Загрузите последнюю версию JAR-файла Spark-коннектора (clickhouse-spark-runtime-3.X_2.X-0.10.X.jar) в S3 бакет.

Предоставьте доступ к S3 бакету

Убедитесь, что задача Glue имеет доступ к этому бакету.

Настройте путь к зависимым JAR-файлам

На вкладке Job details прокрутите страницу вниз, раскройте список Advanced properties и укажите путь к JAR-файлам в поле Dependent JARs path:

Использование AWS Secrets Manager для учетных данных

Вместо того чтобы явно указывать пользователя и пароль ClickHouse в задаче, сохраните их в AWS Secrets Manager и используйте секрет в подключении Glue или в скрипте задачи. Во время выполнения Glue извлекает секрет и добавляет его пары ключ-значение в параметры подключения коннектора.

Создайте секрет

В AWS Secrets Manager создайте секрет типа Другой тип секрета с парами ключ-значение, где ключи совпадают с именами параметров коннектора:

Ключ	Значение
`user`	ваше имя пользователя в ClickHouse
`password`	ваш пароль ClickHouse

Любой ключ, добавленный в секрет, будет передан коннектору, поэтому там также можно хранить host, database или любой другой параметр, если вы не хотите хранить их в коде.

Укажите секрет

Есть два способа передать секрет в задачу.

Вариант 1: привязать его к подключению Glue. При создании или редактировании подключения ClickHouse в Glue Studio укажите в поле AWS secret имя секрета. Любая задача, использующая это подключение, автоматически получает секрет — никаких изменений в коде не требуется.

Вариант 2: передать secretId в параметрах подключения. Используйте этот способ, если секрет не привязан к подключению. Добавьте secretId рядом с connectionName:

source = glueContext.create_dynamic_frame.from_options(
    connection_type="marketplace.spark",
    connection_options={
        "connectionName": "<your-connection-name>",
        "secretId": "clickhouse/glue/credentials",
        "database": "default",
        "table": "example_table"
    },
    transformation_ctx="clickhouse_source"
)

val source = glueContext.getSource(
  connectionType = "marketplace.spark",
  connectionOptions = JsonOptions(Map(
    "connectionName" -> "<your-connection-name>",
    "secretId" -> "clickhouse/glue/credentials",
    "database" -> "default",
    "table" -> "example_table"
  )),
  transformationContext = "clickhouseSource"
)

Ключи user и password из секрета добавляются к параметрам коннектора во время выполнения, поэтому читать их в скрипте не нужно.

Примеры

В примерах ниже используется marketplace.spark, а для ссылки на коннектор — connectionName. Если вы установили коннектор вручную (вкладка Установка вручную), используйте connection_type="custom.spark" и вместо него передавайте className, host, http_port, user и password напрямую в параметре options.

Если вы привязали секрет AWS к самому подключению (вариант 1 в разделе Использование AWS Secrets Manager для учетных данных), уберите secretId из options — Glue автоматически получит учетные данные из подключения.

Вы можете использовать коннектор ClickHouse как источник или целевое назначение в визуальном редакторе Glue Studio. Просто перетащите компонент Spark-коннектора ClickHouse на холст и подключите его к своему конвейеру данных.

Визуальный редактор Glue Studio с коннектором ClickHouse

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.{GlueArgParser, Job, JsonOptions}
import org.apache.spark.SparkContext
import scala.collection.JavaConverters._

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

    val sc = new SparkContext()
    val glueContext = new GlueContext(sc)
    Job.init(args("JOB_NAME"), glueContext, args.asJava)

    val readOptions = JsonOptions(Map(
      "connectionName" -> "<your-connection-name>",
      "secretId" -> "clickhouse/glue/credentials",
      "database" -> "default",
      "table" -> "example_table"
    ))

    val source = glueContext.getSource(
      connectionType = "marketplace.spark",
      connectionOptions = readOptions,
      transformationContext = "clickhouseSource"
    )
    val dyf = source.getDynamicFrame()

    val writeOptions = JsonOptions(Map(
      "connectionName" -> "<your-connection-name>",
      "secretId" -> "clickhouse/glue/credentials",
      "database" -> "default",
      "table" -> "target_table"
    ))

    glueContext.getSink(
      connectionType = "marketplace.spark",
      connectionOptions = writeOptions
    ).writeDynamicFrame(dyf)

    Job.commit()
  }
}

import sys
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

args = getResolvedOptions(sys.argv, ['JOB_NAME'])

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

read_options = {
    "connectionName": "<your-connection-name>",
    "secretId": "clickhouse/glue/credentials",
    "database": "default",
    "table": "example_table"
}

source = glueContext.create_dynamic_frame.from_options(
    connection_type="marketplace.spark",
    connection_options=read_options,
    transformation_ctx="clickhouse_source"
)
dyf = source

logger.info(f"Read {dyf.count()} rows from ClickHouse")

write_options = {
    "connectionName": "<your-connection-name>",
    "secretId": "clickhouse/glue/credentials",
    "database": "default",
    "table": "target_table"
}

glueContext.write_dynamic_frame.from_options(
    frame=dyf,
    connection_type="marketplace.spark",
    connection_options=write_options,
    transformation_ctx="clickhouse_sink"
)

job.commit()

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