تكامل Amazon Glue مع ClickHouse وSpark

تكامل ClickHouse مع Amazon Glue

تُعد Amazon Glue خدمة تكامل بيانات مُدارة بالكامل وبدون خوادم تقدّمها Amazon Web Services ‏(AWS). وهي تُبسّط عملية اكتشاف البيانات وإعدادها وتحويلها لأغراض التحليلات والتعلّم الآلي وتطوير التطبيقات.

​

التثبيت

لدمج شيفرة Glue الخاصة بك مع ClickHouse، يمكنك استخدام موصل Spark الرسمي الخاص بنا في Glue بإحدى الطريقتين التاليتين:

• تثبيت موصل ClickHouse Glue من AWS Marketplace (موصى به).
• إضافة ملفات JAR الخاصة بـ موصل Spark يدويًا إلى مهمة Glue.

AWS Marketplace

1. الاشتراك في الـ موصل

   للوصول إلى الـ موصل في حسابك، اشترك في موصل ClickHouse AWS Glue من AWS Marketplace.

2. منح الأذونات المطلوبة

   تأكد من أن دور IAM الخاص بمهمة Glue لديه الأذونات اللازمة، كما هو موضح في دليل الحد الأدنى من الامتيازات.

3. تفعيل الـ موصل وإنشاء اتصال

   بعد الاشتراك، حدِّد إصدار Glue الذي يطابق متطلبات مهمتك. في قسم Additional details، ضمن Usage instructions، انقر على الرابط Open Glue Studio - Add ClickHouse connector. سيؤدي ذلك إلى فتح صفحة إنشاء اتصال في Glue مع تعبئة الحقول الأساسية مسبقًا. امنح الـ اتصال اسمًا ثم اضغط على create (ولا حاجة إلى إدخال تفاصيل الاتصال الخاصة بـ ClickHouse في هذه المرحلة).

4. الاستخدام في مهمة Glue

   في مهمة Glue، حدِّد علامة التبويب Job details، ثم وسِّع نافذة Advanced properties. ضمن قسم Connections، حدِّد الـ اتصال الذي أنشأته للتو. يضيف الـ موصل تلقائيًا ملفات JAR المطلوبة إلى runtime الخاص بالمهمة.

   تأكد من تحديد إصدار الـ موصل الذي يطابق تكوين مهمة Glue الخاصة بك:

   • Glue 4: Spark 3.3، سكالا 2، بايثون 3
   • Glue 5: Spark 3.5، سكالا 2، بايثون 3

التثبيت اليدوي

لإضافة ملفات JAR المطلوبة يدويًا، يُرجى اتباع ما يلي:

1. رفع ملف JAR الخاص بالـ موصل

   ارفع أحدث ملف JAR لـ موصل Spark (clickhouse-spark-runtime-3.X_2.X-0.10.X.jar) إلى S3 bucket.

2. منح حق الوصول إلى S3 bucket

   تأكد من أن مهمة Glue لديها حق الوصول إلى هذا الـ bucket.

3. تكوين مسار 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 مباشرةً ضمن الخيارات بدلًا من ذلك.

إذا كنت قد أرفقت سرًا من AWS بالاتصال نفسه (الخيار 1 في استخدام AWS Secrets Manager لبيانات الاعتماد)، فاحذف secretId من الخيارات — إذ يستخرج Glue بيانات الاعتماد من الاتصال تلقائيًا.

المحرر المرئي

يمكنك استخدام موصل ClickHouse إما كمصدر أو كهدف في المحرر المرئي لـ Glue Studio. ما عليك سوى سحب مكوّن ClickHouse Spark Connector إلى لوحة العمل وربطه بخط أنابيب البيانات لديك.

سكالا

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.