Amazon Glue를 ClickHouse 및 Spark와 통합

ClickHouse와 Amazon Glue 통합

Amazon Glue는 Amazon Web Services(AWS)에서 제공하는 완전 관리형 서버리스 데이터 통합 서비스입니다. 분석, 머신 러닝, 애플리케이션 개발에 사용할 데이터를 탐색하고, 준비하고, 변환하는 과정을 간소화합니다.

​

설치

Glue 코드와 ClickHouse를 통합하려면 다음 방법 중 하나로 Glue에서 공식 Spark 커넥터를 사용할 수 있습니다.

• AWS Marketplace에서 ClickHouse Glue 커넥터를 설치합니다(권장).
• Spark 커넥터의 JARs를 Glue 작업에 수동으로 추가합니다.

AWS Marketplace

1. 커넥터 구독

   계정에서 커넥터에 액세스하려면 AWS Marketplace에서 ClickHouse AWS Glue Connector를 구독하십시오.

2. 필수 권한 부여

   최소 권한 가이드에 설명된 대로, Glue 작업의 IAM role에 필요한 권한이 있는지 확인하십시오.

3. 커넥터 활성화 및 연결 생성

   구독한 후 작업 요구 사항에 맞는 Glue 버전을 선택하십시오. Additional details 섹션의 Usage instructions 아래에서 Open Glue Studio - Add ClickHouse connector 링크를 클릭하십시오. 그러면 주요 필드가 미리 채워진 Glue 연결 생성 페이지가 열립니다. 연결 이름을 입력한 다음 create를 누르십시오(이 단계에서는 ClickHouse 연결 세부 정보를 입력할 필요가 없습니다).

4. Glue 작업에서 사용

   Glue 작업에서 Job details 탭을 선택한 다음 Advanced properties 창을 펼치십시오. Connections 섹션에서 방금 생성한 연결을 선택하십시오. 커넥터가 필요한 JARs를 작업 런타임에 자동으로 주입합니다.

   Glue 작업 구성과 일치하는 커넥터 버전을 선택해야 합니다.

   • Glue 4: Spark 3.3, Scala 2, Python 3
   • Glue 5: Spark 3.5, Scala 2, Python 3

수동 설치

필요한 JARs를 수동으로 추가하려면 다음 단계를 따르십시오.

1. 커넥터 JAR 업로드

   최신 Spark 커넥터 JAR(clickhouse-spark-runtime-3.X_2.X-0.10.X.jar)을 S3 버킷에 업로드합니다.

2. S3 버킷에 대한 액세스 권한 부여

   Glue 작업이 이 버킷에 액세스할 수 있는지 확인합니다.

3. 종속 JAR 경로 구성

   Job details 탭에서 아래로 스크롤하여 Advanced properties 드롭다운을 펼친 다음 Dependent JARs path에 JAR 경로를 입력합니다.

​

자격 증명에 AWS Secrets Manager 사용

작업에 ClickHouse 사용자와 비밀번호를 하드코딩하는 대신, AWS Secrets Manager에 저장하고 Glue 연결 또는 작업 스크립트에서 해당 시크릿을 참조하십시오. 런타임에 Glue는 시크릿을 가져와 해당 key-value 쌍을 커넥터의 연결 옵션에 병합합니다.

​

시크릿 생성

AWS Secrets Manager에서 Other type of secret 유형의 시크릿을 생성하고, 키가 커넥터의 옵션 이름과 일치하도록 key-value 쌍을 설정하십시오:

Key	Value
user	ClickHouse 사용자 이름
password	ClickHouse 비밀번호

시크릿에 추가한 모든 키는 커넥터로 전달되므로, 코드에 포함하지 않으려면 host, database 또는 다른 옵션도 여기에 저장할 수 있습니다.

​

시크릿 참조

작업에 시크릿을 연결하는 방법은 두 가지입니다.

옵션 1: Glue 연결에 시크릿을 연결합니다. Glue Studio에서 ClickHouse 연결을 생성하거나 편집할 때 AWS secret 필드에 시크릿 이름을 설정하십시오. 이 연결을 사용하는 모든 작업은 시크릿을 자동으로 확인하므로 코드를 변경할 필요가 없습니다.

옵션 2: connection options에 secretId를 전달합니다. 시크릿이 연결에 연결되어 있지 않을 때 이 방법을 사용하십시오. connectionName과 함께 secretId를 추가하십시오.

Python

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"
)

Scala

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가 연결에서 자격 증명을 자동으로 가져옵니다.

비주얼 에디터

ClickHouse 커넥터는 Glue Studio 비주얼 에디터에서 소스 또는 대상으로 사용할 수 있습니다. ClickHouse Spark Connector 컴포넌트를 캔버스로 끌어와 데이터 파이프라인에 연결하면 됩니다.

Scala

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()
  }
}

Python

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 문서를 참조하십시오.