موصل Flink - ClickHouse Documentation

هذا هو موصل sink الرسمي لـ Apache Flink والمدعوم من ClickHouse. وقد بُني باستخدام AsyncSinkBase في Flink وJava client الرسمي لـ ClickHouse. يدعم هذا الموصل واجهة DataStream API في Apache Flink. ومن المخطط دعم Table API في إصدار لاحق.

المتطلبات

Java 11+ (لـ Flink 1.17+) أو 17+ (لـ Flink 2.0+)
Apache Flink 1.17+

مصفوفة توافق إصدارات Flink

ينقسم الموصل إلى حزمتَي artifact لدعم كلٍّ من Flink 1.17+ وFlink 2.0+. اختر الـ artifact المطابق لإصدار Flink الذي تريده:

إصدار Flink	Artifact	إصدار عميل Java لـ ClickHouse	إصدار Java المطلوب
الأحدث	flink-connector-clickhouse-2.0.0	0.9.5	Java 17+
2.0.1	flink-connector-clickhouse-2.0.0	0.9.5	Java 17+
2.0.0	flink-connector-clickhouse-2.0.0	0.9.5	Java 17+
1.20.2	flink-connector-clickhouse-1.17	0.9.5	Java 11+
1.19.3	flink-connector-clickhouse-1.17	0.9.5	Java 11+
1.18.1	flink-connector-clickhouse-1.17	0.9.5	Java 11+
1.17.2	flink-connector-clickhouse-1.17	0.9.5	Java 11+

لم يُختبر هذا الموصل مع إصدارات Flink الأقدم من 1.17.2

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

استيراده كتبعية

بالنسبة إلى Flink 2.0+

Maven
Gradle
SBT

<dependency>
    <groupId>com.clickhouse.flink</groupId>
    <artifactId>flink-connector-clickhouse-2.0.0</artifactId>
    <version>{{ stable_version }}</version>
    <classifier>all</classifier>
</dependency>

dependencies {
    implementation("com.clickhouse.flink:flink-connector-clickhouse-2.0.0:{{ stable_version }}")
}

libraryDependencies += "com.clickhouse.flink" % "flink-connector-clickhouse-2.0.0" % {{ stable_version }} classifier "all"

لإصدار Flink 1.17+

Maven
Gradle
SBT

<dependency>
    <groupId>com.clickhouse.flink</groupId>
    <artifactId>flink-connector-clickhouse-1.17</artifactId>
    <version>{{ stable_version }}</version>
    <classifier>all</classifier>
</dependency>

dependencies {
    implementation("com.clickhouse.flink:flink-connector-clickhouse-1.17:{{ stable_version }}")
}

libraryDependencies += "com.clickhouse.flink" % "flink-connector-clickhouse-1.17" % {{ stable_version }} classifier "all"

تنزيل ملف JAR الثنائي

نمط تسمية ملف JAR الثنائي هو:

flink-connector-clickhouse-${flink_version}-${stable_version}-all.jar

حيث:

flink_version هو أحد الخيارين 2.0.0 أو 1.17
stable_version هو رقم إصدار artifact مستقر

يمكنك العثور على جميع ملفات JAR المُتاحة والصادرة في Maven Central Repository.

استخدام واجهة برمجة تطبيقات DataStream

مقتطف

لنفترض أنك تريد إدراج بيانات CSV خام في ClickHouse:

Java

public static void main(String[] args) {
    // قم بتهيئة ClickHouseClient
    ClickHouseClientConfig clickHouseClientConfig = new ClickHouseClientConfig(url, username, password, database, tableName);

    // أنشئ ElementConverter
    ElementConverter<String, ClickHousePayload> convertorString = new ClickHouseConvertor<>(String.class);

    // أنشئ sink واضبط التنسيق باستخدام `setClickHouseFormat`
    ClickHouseAsyncSink<String> csvSink = new ClickHouseAsyncSink<>(
            convertorString,
            MAX_BATCH_SIZE,
            MAX_IN_FLIGHT_REQUESTS,
            MAX_BUFFERED_REQUESTS,
            MAX_BATCH_SIZE_IN_BYTES,
            MAX_TIME_IN_BUFFER_MS,
            MAX_RECORD_SIZE_IN_BYTES,
            clickHouseClientConfig
    );

    csvSink.setClickHouseFormat(ClickHouseFormat.CSV);

    // أخيرًا، صِل DataStream الخاصة بك بـ sink.
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    Path csvFilePath = new Path(fileFullName);
    FileSource<String> csvSource = FileSource
            .forRecordStreamFormat(new TextLineInputFormat(), csvFilePath)
            .build();

    env.fromSource(
            csvSource,
            WatermarkStrategy.noWatermarks(),
            "GzipCsvSource"
    ).sinkTo(csvSink);
}

يمكنك العثور على المزيد من الأمثلة والمقتطفات في اختباراتنا:

مثال للبدء السريع

أنشأنا مثالًا قائمًا على Maven لتسهيل البدء باستخدام ClickHouse Sink:

للحصول على إرشادات أكثر تفصيلًا، راجع دليل الأمثلة

خيارات الاتصال بواجهة برمجة تطبيقات DataStream

خيارات عميل ClickHouse

المعلمات	الوصف	القيمة الافتراضية	مطلوب
`url`	عنوان URL المؤهل بالكامل لـ ClickHouse	غير متاح	نعم
`username`	اسم مستخدم قاعدة بيانات ClickHouse	غير متاح	نعم
`password`	كلمة مرور قاعدة بيانات ClickHouse	غير متاح	نعم
`database`	اسم قاعدة بيانات ClickHouse	غير متاح	نعم
`table`	اسم جدول ClickHouse	غير متاح	نعم
`options`	خريطة خيارات تهيئة عميل Java	خريطة فارغة	لا
`serverSettings`	خريطة إعدادات جلسة خادم ClickHouse	خريطة فارغة	لا
`enableJsonSupportAsString`	إعداد في خادم ClickHouse لتوقّع قيمة `String` منسّقة بتنسيق JSON من أجل نوع بيانات JSON	true	لا

يجب تمرير options وserverSettings إلى العميل بصيغة Map<String, String>. وسيؤدي استخدام خريطة فارغة لأيٍّ منهما إلى استخدام الإعدادات الافتراضية للعميل أو الخادم، على الترتيب.

جميع خيارات عميل Java المتاحة مُدرجة في ClientConfigProperties.java وصفحة التوثيق هذه.جميع إعدادات جلسة الخادم المتاحة مُدرجة في صفحة التوثيق هذه.

على سبيل المثال:

Java

Map<String, String> javaClientOptions = Map.of(
    ClientConfigProperties.CA_CERTIFICATE.getKey(), "<my_CA_cert>",
    ClientConfigProperties.SSL_CERTIFICATE.getKey(), "<my_SSL_cert>",
    ClientConfigProperties.CLIENT_NETWORK_BUFFER_SIZE.getKey(), "30000",
    ClientConfigProperties.HTTP_MAX_OPEN_CONNECTIONS.getKey(), "5"
);

Map<String, String> serverSettings = Map.of(
    "insert_deduplicate", "1"
);

ClickHouseClientConfig clickHouseClientConfig = new ClickHouseClientConfig(
    url,
    username,
    password,
    database,
    tableName,
    javaClientOptions,
    serverSettings,
    false // تمكين enableJsonSupportAsString
);

خيارات الـ sink

تأتي الخيارات التالية مباشرةً من AsyncSinkBase في Flink:

المعلمات	الوصف	القيمة الافتراضية	مطلوب
`maxBatchSize`	الحد الأقصى لعدد السجلات التي يمكن إدراجها في دفعة واحدة	N/A	نعم
`maxInFlightRequests`	الحد الأقصى لعدد الطلبات قيد التنفيذ المسموح بها قبل أن يطبّق الـ sink ضغطًا عكسيًا	N/A	نعم
`maxBufferedRequests`	الحد الأقصى لعدد السجلات التي يمكن تخزينها مؤقتًا في الـ sink قبل تطبيق الضغط العكسي	N/A	نعم
`maxBatchSizeInBytes`	الحد الأقصى لحجم الدفعة (بالبايت). ستكون جميع الدفعات المرسلة أصغر من هذا الحجم أو مساوية له	N/A	نعم
`maxTimeInBufferMS`	الحد الأقصى للمدة التي يمكن أن يبقى فيها السجل في الـ sink قبل تفريغه	N/A	نعم
`maxRecordSizeInBytes`	الحد الأقصى لحجم السجل الذي يقبله الـ sink، وستُرفض تلقائيًا أي سجلات أكبر من ذلك	N/A	نعم

أنواع البيانات المدعومة

يوفّر الجدول أدناه مرجعًا سريعًا لتحويل أنواع البيانات عند الإدراج من Flink إلى ClickHouse.

إدراج البيانات من Flink في ClickHouse

نوع Java	نوع ClickHouse	مدعوم	طريقة التسلسل
`byte`/`Byte`	`Int8`	✅	`DataWriter.writeInt8`
`short`/`Short`	`Int16`	✅	`DataWriter.writeInt16`
`int`/`Integer`	`Int32`	✅	`DataWriter.writeInt32`
`long`/`Long`	`Int64`	✅	`DataWriter.writeInt64`
`BigInteger`	`Int128`	✅	`DataWriter.writeInt128`
`BigInteger`	`Int256`	✅	`DataWriter.writeInt256`
`short`/`Short`	`UInt8`	✅	`DataWriter.writeUInt8`
`int`/`Integer`	`UInt8`	✅	`DataWriter.writeUInt8`
`int`/`Integer`	`UInt16`	✅	`DataWriter.writeUInt16`
`long`/`Long`	`UInt32`	✅	`DataWriter.writeUInt32`
`long`/`Long`	`UInt64`	✅	`DataWriter.writeUInt64`
`BigInteger`	`UInt64`	✅	`DataWriter.writeUInt64`
`BigInteger`	`UInt128`	✅	`DataWriter.writeUInt128`
`BigInteger`	`UInt256`	✅	`DataWriter.writeUInt256`
`BigDecimal`	`Decimal`	✅	`DataWriter.writeDecimal`
`BigDecimal`	`Decimal32`	✅	`DataWriter.writeDecimal`
`BigDecimal`	`Decimal64`	✅	`DataWriter.writeDecimal`
`BigDecimal`	`Decimal128`	✅	`DataWriter.writeDecimal`
`BigDecimal`	`Decimal256`	✅	`DataWriter.writeDecimal`
`float`/`Float`	`Float`	✅	`DataWriter.writeFloat32`
`double`/`Double`	`Double`	✅	`DataWriter.writeFloat64`
`boolean`/`Boolean`	`Boolean`	✅	`DataWriter.writeBoolean`
`String`	`String`	✅	`DataWriter.writeString`
`String`	`FixedString`	✅	`DataWriter.writeFixedString`
`LocalDate`	`Date`	✅	`DataWriter.writeDate`
`LocalDate`	`Date32`	✅	`DataWriter.writeDate32`
`LocalDateTime`	`DateTime`	✅	`DataWriter.writeDateTime`
`ZonedDateTime`	`DateTime`	✅	`DataWriter.writeDateTime`
`LocalDateTime`	`DateTime64`	✅	`DataWriter.writeDateTime64`
`ZonedDateTime`	`DateTime64`	✅	`DataWriter.writeDateTime64`
`int`/`Integer`	`Time`	❌	غير متاح
`long`/`Long`	`Time64`	❌	غير متاح
`byte`/`Byte`	`Enum8`	✅	`DataWriter.writeInt8`
`int`/`Integer`	`Enum16`	✅	`DataWriter.writeInt16`
`java.util.UUID`	`UUID`	✅	`DataWriter.writeIntUUID`
`String`	`JSON`	✅	`DataWriter.writeJSON`
`Array<Type>`	`Array<Type>`	✅	`DataWriter.writeArray`
`Map<K,V>`	`Map<K,V>`	✅	`DataWriter.writeMap`
`Tuple<Type,..>`	`Tuple<T1,T2,..>`	✅	`DataWriter.writeTuple`
`Object`	`Variant`	❌	غير متاح

ملاحظات:

يجب توفير ZoneId عند إجراء عمليات على التاريخ.
يجب توفير الدقة والمقياس عند إجراء عمليات على القيم العشرية.
لكي يتمكن ClickHouse من تحليل String في Java على أنه JSON، يجب تمكين enableJsonSupportAsString في ClickHouseClientConfig.
يتطلب الموصّل ElementConvertor لربط العناصر في DataStream المدخل بحمولات ClickHouse. ولهذا الغرض، يوفّر الموصّل ClickHouseConvertor وPOJOConvertor، ويمكنك استخدامهما لتنفيذ هذا الربط باستخدام طرق التسلسل الخاصة بـ DataWriter المذكورة أعلاه.

تنسيقات الإدخال المدعومة

يمكنك العثور على قائمة تنسيقات الإدخال المتاحة في ClickHouse في صفحة التوثيق هذه وClickHouseFormat.java. لتحديد التنسيق الذي ينبغي أن يستخدمه الموصل لتحويل DataStream إلى ClickHouse حمولات، استخدم الدالة setClickHouseFormat. على سبيل المثال:

ClickHouseAsyncSink<String> csvSink = new ClickHouseAsyncSink<>(
        convertorString,
        MAX_BATCH_SIZE,
        MAX_IN_FLIGHT_REQUESTS,
        MAX_BUFFERED_REQUESTS,
        MAX_BATCH_SIZE_IN_BYTES,
        MAX_TIME_IN_BUFFER_MS,
        MAX_RECORD_SIZE_IN_BYTES,
        clickHouseClientConfig
);
csvSink.setClickHouseFormat(ClickHouseFormat.CSV);

افتراضيًا، سيستخدم الموصل تنسيق RowBinaryWithDefaults أو RowBinary إذا ضُبطت القيمة setSupportDefault في ClickHouseClientConfig صراحةً على true أو false، على الترتيب.

المقاييس

يوفّر الموصل المقاييس الإضافية التالية بالإضافة إلى مقاييس Flink الحالية:

المقياس	الوصف	النوع	الحالة
`numBytesSend`	العدد الإجمالي للبايتات المُرسلة إلى ClickHouse ضمن حمولة الطلب. ملاحظة: يقيس هذا المقياس حجم البيانات المُسلسلة المُرسلة عبر الشبكة، وقد يختلف عن `written_bytes` في `system.query_log` في ClickHouse، الذي يعكس عدد البايتات الفعلي المكتوب إلى وحدة التخزين بعد المعالجة	عداد	✅
`numRecordSend`	العدد الإجمالي للسجلات المُرسلة إلى ClickHouse	عداد	✅
`numRequestSubmitted`	العدد الإجمالي للطلبات المُرسلة (العدد الفعلي لعمليات التفريغ المنفَّذة)	عداد	✅
`numOfDroppedBatches`	العدد الإجمالي للدفعات التي أُسقطت بسبب إخفاقات غير قابلة لإعادة المحاولة	عداد	✅
`numOfDroppedRecords`	العدد الإجمالي للسجلات التي أُسقطت بسبب إخفاقات غير قابلة لإعادة المحاولة	عداد	✅
`totalBatchRetries`	العدد الإجمالي لمحاولات إعادة الدفعات بسبب إخفاقات قابلة لإعادة المحاولة	عداد	✅
`writeLatencyHistogram`	مُدرَّج تكراري لتوزيع زمن استجابة عمليات الكتابة الناجحة (مللي ثانية)	مُدرَّج تكراري	✅
`writeFailureLatencyHistogram`	مُدرَّج تكراري لتوزيع زمن استجابة عمليات الكتابة الفاشلة (مللي ثانية)	مُدرَّج تكراري	✅
`triggeredByMaxBatchSizeCounter`	العدد الإجمالي لعمليات التفريغ الناتجة عن بلوغ `maxBatchSize`	عداد	✅
`triggeredByMaxBatchSizeInBytesCounter`	العدد الإجمالي لعمليات التفريغ الناتجة عن بلوغ `maxBatchSizeInBytes`	عداد	✅
`triggeredByMaxTimeInBufferMSCounter`	العدد الإجمالي لعمليات التفريغ الناتجة عن بلوغ `maxTimeInBufferMS`	عداد	✅
`actualRecordsPerBatch`	مُدرَّج تكراري لتوزيع العدد الفعلي للسجلات في كل دفعة	مُدرَّج تكراري	✅
`actualBytesPerBatch`	مُدرَّج تكراري لتوزيع العدد الفعلي للبايتات في كل دفعة	مُدرَّج تكراري	✅

القيود

يوفّر sink حاليًا ضمان تسليم مرة واحدة على الأقل. ويجري تتبّع العمل لتحقيق exactly-once semantics هنا.
لا يدعم sink بعدُ قائمة انتظار الرسائل الميتة (DLQ) لتخزين السجلات غير القابلة للمعالجة مؤقتًا. وحتى ذلك الحين، سيحاول الموصل إعادة إدراج السجلات التي تفشل، وسيتخلّص منها إذا لم ينجح ذلك. ويجري تتبّع هذه الميزة هنا.
لا يدعم sink بعدُ الإنشاء عبر Table API الخاصة بـ Flink أو Flink SQL. ويجري تتبّع هذه الميزة هنا.

توافق إصدارات ClickHouse والأمان

يُختبَر موصل مع مجموعة من إصدارات ClickHouse الحديثة، بما في ذلك latest وhead، عبر سير عمل CI يومي. وتُحدَّث الإصدارات المختبَرة دوريًا مع اعتماد إصدارات ClickHouse الجديدة. اطّلع هنا على الإصدارات التي يُختبَر موصل عليها يوميًا.
راجع سياسة أمان ClickHouse للتعرّف على الثغرات الأمنية المعروفة وكيفية الإبلاغ عن أي ثغرة.
نوصي بترقية موصل باستمرار حتى لا تفوتك الإصلاحات الأمنية والتحسينات الجديدة.
إذا واجهت مشكلة في الترحيل، فيُرجى إنشاء issue على GitHub وسنرد عليك!

الاستخدام المتقدم والموصى به

للحصول على أفضل أداء، تأكد من أن نوع العنصر في DataStream لديك ليس من النوع Generic - راجع هذا الشرح لتمييز الأنواع في Flink. فالعناصر غير العامة تتجنب كلفة التسلسل الإضافية التي يفرضها Kryo وتُحسّن معدل النقل إلى ClickHouse.
نوصي بضبط maxBatchSize على 1000 كحد أدنى، ويفضَّل أن يكون بين 10,000 و100,000. راجع هذا الدليل حول عمليات الإدراج المجمّعة لمزيد من المعلومات.
لإجراء إزالة التكرار أو upsert إلى ClickHouse بأسلوب OLTP، راجع صفحة التوثيق هذه. ملاحظة: لا تخلط بين هذا وبين إزالة التكرار على مستوى الدُفعات التي تحدث عند إعادة المحاولة.

استكشاف الأخطاء وإصلاحها

CANNOT_READ_ALL_DATA

قد يظهر الخطأ التالي:

com.clickhouse.client.api.ServerException: Code: 33. DB::Exception: Cannot read all data. Bytes read: 9205. Bytes expected: 1100022.: (at row 9) : While executing BinaryRowInputFormat. (CANNOT_READ_ALL_DATA)

السبب: في معظم الحالات، يعني الخطأ CANNOT_READ_ALL_DATA أن مخطط جدول ClickHouse لديك لم يعد متطابقًا مع مخطط سجل Flink. وقد يحدث ذلك عندما يُعدَّل أحدهما أو كلاهما بطريقة غير متوافقة مع الإصدارات السابقة. الحل: حدِّث المخطط في جدول ClickHouse لديك أو نوع بيانات إدخال الموصل (أو كليهما) بحيث يصبحان متوافقين. وإذا لزم الأمر، فارجع إلى تعيين الأنواع لمعرفة كيفية ربط أنواع Java بأنواع ClickHouse. ملاحظة: إذا كانت لا تزال هناك سجلات قيد النقل، فستحتاج إلى إعادة تعيين حالة Flink عند إعادة تشغيل الموصل.

انخفاض معدل النقل

قد تلاحظ أن معدل نقل الموصل لا يزداد بما يتناسب مع توازي المهمة (عدد مهام Flink) عند الكتابة إلى ClickHouse. السبب: قد تؤدي عملية دمج الأجزاء في الخلفية في ClickHouse إلى إبطاء عمليات الإدراج. يمكن أن يحدث ذلك عندما يكون حجم الدفعة المُعدّ صغيرًا جدًا، أو عندما يقوم الموصل بعملية التفريغ بشكل متكرر جدًا، أو بسبب الجمع بين الأمرين. الحل: راقب المقياسين numRequestSubmitted وactualRecordsPerBatch للمساعدة في تحديد كيفية ضبط حجم الدفعة (maxBatchSize) وعدد مرات التفريغ. راجع أيضًا الاستخدام المتقدم والموصى به للاطلاع على توصيات بشأن حجم الدفعات.

هناك صفوف مفقودة في جدول ClickHouse الخاص بي

السبب: أُسقِطت الدفعة (أو الدفعات) إما بسبب فشل غير قابل لإعادة المحاولة، أو لعدم إمكانية إدراجها ضمن عدد محاولات إعادة المحاولة المُعدّ (يمكن ضبطه عبر ClickHouseClientConfig.setNumberOfRetries()). ملاحظة: يحاول الموصل، افتراضيًا، إعادة إدراج الدفعة حتى 3 مرات قبل إسقاطها. الحل: افحص سجلات TaskManager و/أو تتبّع المكدس لتحديد السبب الجذري.

المساهمة والدعم

إذا كنت ترغب في المساهمة في المشروع أو الإبلاغ عن أي مشكلات، فنحن نرحّب بمشاركتك! تفضّل بزيارة مستودع GitHub الخاص بنا لفتح issue أو اقتراح تحسينات أو إرسال pull request. نرحّب بالمساهمات! يُرجى الاطلاع على دليل المساهمة في المستودع قبل البدء. شكرًا لمساعدتك في تحسين موصل ClickHouse لـ Flink!

​المتطلبات

​مصفوفة توافق إصدارات Flink

​التثبيت والإعداد

​استيراده كتبعية

​بالنسبة إلى Flink 2.0+

​لإصدار Flink 1.17+

​تنزيل ملف JAR الثنائي

​استخدام واجهة برمجة تطبيقات DataStream

​مقتطف

​مثال للبدء السريع

​خيارات الاتصال بواجهة برمجة تطبيقات DataStream

​خيارات عميل ClickHouse

​خيارات الـ sink

​أنواع البيانات المدعومة

​إدراج البيانات من Flink في ClickHouse

​تنسيقات الإدخال المدعومة

​المقاييس

​القيود

​توافق إصدارات ClickHouse والأمان

​الاستخدام المتقدم والموصى به

​استكشاف الأخطاء وإصلاحها

​CANNOT_READ_ALL_DATA

​انخفاض معدل النقل

​هناك صفوف مفقودة في جدول ClickHouse الخاص بي

​المساهمة والدعم

المتطلبات

مصفوفة توافق إصدارات Flink

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

استيراده كتبعية

بالنسبة إلى Flink 2.0+

لإصدار Flink 1.17+

تنزيل ملف JAR الثنائي

استخدام واجهة برمجة تطبيقات DataStream

مقتطف

مثال للبدء السريع

خيارات الاتصال بواجهة برمجة تطبيقات DataStream

خيارات عميل ClickHouse

خيارات الـ sink

أنواع البيانات المدعومة

إدراج البيانات من Flink في ClickHouse

تنسيقات الإدخال المدعومة

المقاييس

القيود

توافق إصدارات ClickHouse والأمان

الاستخدام المتقدم والموصى به

استكشاف الأخطاء وإصلاحها

CANNOT_READ_ALL_DATA

انخفاض معدل النقل

هناك صفوف مفقودة في جدول ClickHouse الخاص بي

المساهمة والدعم