集成 Amazon Glue、ClickHouse 与 Spark

支持 ClickHouse

Amazon Glue 是 Amazon Web Services (AWS) 提供的一项全托管、无服务器数据集成服务。它简化了为分析、机器学习和应用开发而发现、准备和转换数据的过程。

安装

要将你的 Glue 代码与 ClickHouse 集成，可以通过以下任一方式在 Glue 中使用我们的官方 Spark Connector：

从 AWS Marketplace 安装 ClickHouse Glue 连接器 (推荐) 。
手动将 Spark Connector 的 JAR 添加到你的 Glue 作业中。

授予所需权限

确保你的 Glue 作业所使用的 IAM role 具备所需权限，具体请参见最低权限指南。

激活连接器并创建连接

订阅后，选择与你的作业要求匹配的 Glue 版本。在 Additional details 部分的 Usage instructions 下，点击 Open Glue Studio - Add ClickHouse connector 的链接。该链接会打开 Glue 连接创建页面，并预先填好关键字段。为连接命名后，点击 create 即可 (此阶段无需提供 ClickHouse 连接信息) 。

AWS Marketplace 中 ClickHouse Glue 连接器的使用说明

在 Glue 作业中使用

在你的 Glue 作业中，选择 Job details 选项卡，然后展开 Advanced properties 窗口。在 Connections 部分中，选择你刚刚创建的连接。该连接器会自动将所需 JAR 注入作业运行时。

如需手动添加所需的 JAR，请按以下步骤操作：

上传连接器 JAR

将最新的 Spark Connector JAR (clickhouse-spark-runtime-3.X_2.X-0.10.X.jar) 上传到一个 S3 bucket。

授予对 S3 bucket 的访问权限

确保 Glue 作业可以访问该 bucket。

配置依赖 JAR 路径

在 Job details 选项卡下，向下滚动并展开 Advanced properties 下拉菜单，然后在 Dependent JARs path 中填写 JAR 路径：

使用 AWS Secrets Manager 管理凭证

不要在作业中硬编码 ClickHouse 用户名和密码，而应将其存储在 AWS Secrets Manager 中，并在 Glue 连接或作业脚本中引用该密钥。在运行时，Glue 会拉取该密钥，并将其中的键值对合并到连接器的连接选项中。

创建密钥

在 AWS Secrets Manager 中，创建一个类型为 其他类型的密钥 的密钥，并添加键值对，其中键名需与 connector 的选项名称一致：

键	值
`user`	你的 ClickHouse 用户名
`password`	你的 ClickHouse 密码

你在该密钥中放入的任何键都会被转发给 connector，因此如果你想避免将它们写在代码中，也可以在其中存储 host、database 或任何其他选项。

引用密钥

有两种方式可让作业使用该密钥。

选项 1：将其附加到 Glue 连接。 在 Glue Studio 中创建或编辑 ClickHouse 连接时，将 AWS 密钥 字段设置为该密钥的名称。任何使用此连接的作业都会自动解析该密钥，无需修改代码。

选项 2：在连接选项中传入 secretId。 当密钥未附加到连接时，请使用此方法。在 connectionName 旁边添加 secretId：

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。

如果你将 AWS 密钥附加到连接本身 (使用 AWS Secrets Manager 管理凭据中的选项 1) ，请从选项中去掉 secretId——Glue 会自动从连接中解析凭据。

你可以在 Glue Studio 可视化编辑器中将 ClickHouse 连接器用作数据源或目标端。只需将 ClickHouse Spark 连接器组件拖到画布上，然后连接到你的数据管道即可。

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 文档。