` في إعدادات Spark:
```text theme={null}
spark.sql.catalog.clickhouse.option.allow_nullable_key 1
spark.sql.catalog.clickhouse.option.index_granularity 8192
```
أو اضبطها عند إنشاء الجداول باستخدام Spark SQL:
```sql theme={null}
CREATE TABLE clickhouse.default.my_table (
id INT,
name STRING
) USING ClickHouse
TBLPROPERTIES (
engine = 'MergeTree()',
order_by = 'id',
'settings.allow_nullable_key' = '1',
'settings.index_granularity' = '8192'
)
```
## إعدادات ClickHouse Cloud
عند الاتصال بـ [ClickHouse Cloud](https://clickhouse.com)، تأكد من تمكين SSL وضبط وضع SSL المناسب. على سبيل المثال:
```text theme={null}
spark.sql.catalog.clickhouse.option.ssl true
spark.sql.catalog.clickhouse.option.ssl_mode NONE
```
## قراءة البيانات
```java theme={null}
public static void main(String[] args) {
// إنشاء جلسة Spark
SparkSession spark = SparkSession.builder()
.appName("example")
.master("local[*]")
.config("spark.sql.catalog.clickhouse", "com.clickhouse.spark.ClickHouseCatalog")
.config("spark.sql.catalog.clickhouse.host", "127.0.0.1")
.config("spark.sql.catalog.clickhouse.protocol", "http")
.config("spark.sql.catalog.clickhouse.http_port", "8123")
.config("spark.sql.catalog.clickhouse.user", "default")
.config("spark.sql.catalog.clickhouse.password", "123456")
.config("spark.sql.catalog.clickhouse.database", "default")
.config("spark.clickhouse.write.format", "json")
.getOrCreate();
Dataset df = spark.sql("select * from clickhouse.default.example_table");
df.show();
spark.stop();
}
```
```java theme={null}
object NativeSparkRead extends App {
val spark = SparkSession.builder
.appName("example")
.master("local[*]")
.config("spark.sql.catalog.clickhouse", "com.clickhouse.spark.ClickHouseCatalog")
.config("spark.sql.catalog.clickhouse.host", "127.0.0.1")
.config("spark.sql.catalog.clickhouse.protocol", "http")
.config("spark.sql.catalog.clickhouse.http_port", "8123")
.config("spark.sql.catalog.clickhouse.user", "default")
.config("spark.sql.catalog.clickhouse.password", "123456")
.config("spark.sql.catalog.clickhouse.database", "default")
.config("spark.clickhouse.write.format", "json")
.getOrCreate
val df = spark.sql("select * from clickhouse.default.example_table")
df.show()
spark.stop()
}
```
```python theme={null}
from pyspark.sql import SparkSession
packages = [
"com.clickhouse.spark:clickhouse-spark-runtime-3.4_2.12:0.8.0",
"com.clickhouse:clickhouse-client:0.7.0",
"com.clickhouse:clickhouse-http-client:0.7.0",
"org.apache.httpcomponents.client5:httpclient5:5.2.1"
]
spark = (SparkSession.builder
.config("spark.jars.packages", ",".join(packages))
.getOrCreate())
spark.conf.set("spark.sql.catalog.clickhouse", "com.clickhouse.spark.ClickHouseCatalog")
spark.conf.set("spark.sql.catalog.clickhouse.host", "127.0.0.1")
spark.conf.set("spark.sql.catalog.clickhouse.protocol", "http")
spark.conf.set("spark.sql.catalog.clickhouse.http_port", "8123")
spark.conf.set("spark.sql.catalog.clickhouse.user", "default")
spark.conf.set("spark.sql.catalog.clickhouse.password", "123456")
spark.conf.set("spark.sql.catalog.clickhouse.database", "default")
spark.conf.set("spark.clickhouse.write.format", "json")
df = spark.sql("select * from clickhouse.default.example_table")
df.show()
```
```sql theme={null}
CREATE TEMPORARY VIEW jdbcTable
USING org.apache.spark.sql.jdbc
OPTIONS (
url "jdbc:ch://localhost:8123/default",
dbtable "schema.tablename",
user "username",
password "password",
driver "com.clickhouse.jdbc.ClickHouseDriver"
);
SELECT * FROM jdbcTable;
```
## كتابة البيانات
**استبدال الأقسام غير مدعوم**: لا تدعم واجهة برمجة تطبيقات Catalog حاليًا عمليات الاستبدال على مستوى التقسيم (مثل وضع `overwrite` مع `partitionBy`). هذه الميزة قيد التطوير حاليًا. راجع [مشكلة GitHub رقم 34](https://github.com/ClickHouse/spark-clickhouse-connector/issues/34) لمتابعة هذه الميزة.
```java theme={null}
public static void main(String[] args) throws AnalysisException {
// إنشاء جلسة Spark
SparkSession spark = SparkSession.builder()
.appName("example")
.master("local[*]")
.config("spark.sql.catalog.clickhouse", "com.clickhouse.spark.ClickHouseCatalog")
.config("spark.sql.catalog.clickhouse.host", "127.0.0.1")
.config("spark.sql.catalog.clickhouse.protocol", "http")
.config("spark.sql.catalog.clickhouse.http_port", "8123")
.config("spark.sql.catalog.clickhouse.user", "default")
.config("spark.sql.catalog.clickhouse.password", "123456")
.config("spark.sql.catalog.clickhouse.database", "default")
.config("spark.clickhouse.write.format", "json")
.getOrCreate();
// تحديد schema الخاصة بـ DataFrame
StructType schema = new StructType(new StructField[]{
DataTypes.createStructField("id", DataTypes.IntegerType, false),
DataTypes.createStructField("name", DataTypes.StringType, false),
});
List data = Arrays.asList(
RowFactory.create(1, "Alice"),
RowFactory.create(2, "Bob")
);
// إنشاء DataFrame
Dataset df = spark.createDataFrame(data, schema);
df.writeTo("clickhouse.default.example_table").append();
spark.stop();
}
```
```java theme={null}
object NativeSparkWrite extends App {
// إنشاء جلسة Spark
val spark: SparkSession = SparkSession.builder
.appName("example")
.master("local[*]")
.config("spark.sql.catalog.clickhouse", "com.clickhouse.spark.ClickHouseCatalog")
.config("spark.sql.catalog.clickhouse.host", "127.0.0.1")
.config("spark.sql.catalog.clickhouse.protocol", "http")
.config("spark.sql.catalog.clickhouse.http_port", "8123")
.config("spark.sql.catalog.clickhouse.user", "default")
.config("spark.sql.catalog.clickhouse.password", "123456")
.config("spark.sql.catalog.clickhouse.database", "default")
.config("spark.clickhouse.write.format", "json")
.getOrCreate
// تحديد schema الخاصة بـ DataFrame
val rows = Seq(Row(1, "John"), Row(2, "Doe"))
val schema = List(
StructField("id", DataTypes.IntegerType, nullable = false),
StructField("name", StringType, nullable = true)
)
// إنشاء df
val df: DataFrame = spark.createDataFrame(
spark.sparkContext.parallelize(rows),
StructType(schema)
)
df.writeTo("clickhouse.default.example_table").append()
spark.stop()
}
```
```python theme={null}
from pyspark.sql import SparkSession
from pyspark.sql import Row
# يمكنك استخدام أي مجموعة أخرى من الحزم ما دامت تستوفي Compatibility Matrix المذكورة أعلاه.
packages = [
"com.clickhouse.spark:clickhouse-spark-runtime-3.4_2.12:0.8.0",
"com.clickhouse:clickhouse-client:0.7.0",
"com.clickhouse:clickhouse-http-client:0.7.0",
"org.apache.httpcomponents.client5:httpclient5:5.2.1"
]
spark = (SparkSession.builder
.config("spark.jars.packages", ",".join(packages))
.getOrCreate())
spark.conf.set("spark.sql.catalog.clickhouse", "com.clickhouse.spark.ClickHouseCatalog")
spark.conf.set("spark.sql.catalog.clickhouse.host", "127.0.0.1")
spark.conf.set("spark.sql.catalog.clickhouse.protocol", "http")
spark.conf.set("spark.sql.catalog.clickhouse.http_port", "8123")
spark.conf.set("spark.sql.catalog.clickhouse.user", "default")
spark.conf.set("spark.sql.catalog.clickhouse.password", "123456")
spark.conf.set("spark.sql.catalog.clickhouse.database", "default")
spark.conf.set("spark.clickhouse.write.format", "json")
# إنشاء DataFrame
data = [Row(id=11, name="John"), Row(id=12, name="Doe")]
df = spark.createDataFrame(data)
# كتابة DataFrame إلى ClickHouse
df.writeTo("clickhouse.default.example_table").append()
```
```sql theme={null}
-- resultTable هو DataFrame الوسيط في Spark الذي نريد إدراجه في clickhouse.default.example_table
INSERT INTO TABLE clickhouse.default.example_table
SELECT * FROM resultTable;
```
## عمليات DDL
يمكنك تنفيذ عمليات DDL على مثيل ClickHouse لديك باستخدام Spark SQL، مع حفظ جميع التغييرات فورًا في
ClickHouse.
يتيح لك Spark SQL كتابة الاستعلامات تمامًا كما تفعل في ClickHouse،
لذا يمكنك تنفيذ أوامر مثل CREATE TABLE وTRUNCATE وغيرها مباشرةً، من دون أي تعديل، على سبيل المثال:
عند استخدام Spark SQL، لا يمكن تنفيذ سوى statement واحدة في كل مرة.
```sql theme={null}
USE clickhouse;
```
```sql theme={null}
CREATE TABLE test_db.tbl_sql (
create_time TIMESTAMP NOT NULL,
m INT NOT NULL COMMENT 'part key',
id BIGINT NOT NULL COMMENT 'sort key',
value STRING
) USING ClickHouse
PARTITIONED BY (m)
TBLPROPERTIES (
engine = 'MergeTree()',
order_by = 'id',
settings.index_granularity = 8192
);
```
توضح الأمثلة أعلاه استعلامات Spark SQL، ويمكنك تشغيلها داخل تطبيقك باستخدام أي واجهة برمجة تطبيقات، مثل Java أو Scala
أو PySpark أو shell.
## العمل مع VariantType
يتوفر دعم VariantType في Spark 4.0+، ويتطلب ClickHouse 25.3+ مع تمكين نوعَي JSON/Variant التجريبيين.
يدعم الموصل `VariantType` في Spark للعمل مع البيانات شبه المهيكلة. ويرتبط VariantType بالنوعين `JSON` و`Variant` في ClickHouse، ما يتيح لك تخزين البيانات ذات المخطط المرن والاستعلام عنها بكفاءة.
يركّز هذا القسم تحديدًا على ربط VariantType واستخدامه. للحصول على نظرة عامة كاملة على جميع أنواع البيانات المدعومة، راجع قسم [أنواع البيانات المدعومة](#supported-data-types).
### تعيين أنواع ClickHouse
| نوع ClickHouse | نوع Spark | الوصف |
| ---------------------- | ------------- | ----------------------------------------------------------------- |
| `JSON` | `VariantType` | يخزّن كائنات JSON فقط (يجب أن يبدأ بـ `{`) |
| `Variant(T1, T2, ...)` | `VariantType` | يخزّن أنواعًا متعددة، بما في ذلك الأنواع الأولية والمصفوفات وJSON |
### قراءة بيانات VariantType
عند القراءة من ClickHouse، تُعيَّن الأعمدة `JSON` و`Variant` تلقائيًا إلى `VariantType` في Spark:
```scala theme={null}
// قراءة عمود JSON كـ VariantType
val df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
// الوصول إلى بيانات Variant
df.show()
// تحويل Variant إلى سلسلة JSON للفحص
import org.apache.spark.sql.functions._
df.select(
col("id"),
to_json(col("data")).as("data_json")
).show()
```
```python theme={null}
# قراءة عمود JSON كـ VariantType
df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
# الوصول إلى بيانات Variant
df.show()
# تحويل Variant إلى سلسلة JSON للفحص
from pyspark.sql.functions import to_json
df.select(
"id",
to_json("data").alias("data_json")
).show()
```
```java theme={null}
// قراءة عمود JSON كـ VariantType
Dataset df = spark.sql("SELECT id, data FROM clickhouse.default.json_table");
// الوصول إلى بيانات Variant
df.show();
// تحويل Variant إلى سلسلة JSON للفحص
import static org.apache.spark.sql.functions.*;
df.select(
col("id"),
to_json(col("data")).as("data_json")
).show();
```
### كتابة بيانات من النوع VariantType
يمكنك كتابة بيانات VariantType إلى ClickHouse باستخدام نوعَي الأعمدة JSON أو Variant:
```scala theme={null}
import org.apache.spark.sql.functions._
// إنشاء DataFrame يحتوي على بيانات JSON
val jsonData = Seq(
(1, """{"name": "Alice", "age": 30}"""),
(2, """{"name": "Bob", "age": 25}"""),
(3, """{"name": "Charlie", "city": "NYC"}""")
).toDF("id", "json_string")
// تحليل سلاسل JSON إلى VariantType
val variantDF = jsonData.select(
col("id"),
parse_json(col("json_string")).as("data")
)
// الكتابة إلى ClickHouse باستخدام نوع JSON (كائنات JSON فقط)
variantDF.writeTo("clickhouse.default.user_data").create()
// أو حدِّد Variant بعدة أنواع
spark.sql("""
CREATE TABLE clickhouse.default.mixed_data (
id INT,
data VARIANT
) USING clickhouse
TBLPROPERTIES (
'clickhouse.column.data.variant_types' = 'String, Int64, Bool, JSON',
'engine' = 'MergeTree()',
'order_by' = 'id'
)
""")
```
```python theme={null}
from pyspark.sql.functions import parse_json
# إنشاء DataFrame يحتوي على بيانات JSON
json_data = [
(1, '{"name": "Alice", "age": 30}'),
(2, '{"name": "Bob", "age": 25}'),
(3, '{"name": "Charlie", "city": "NYC"}')
]
df = spark.createDataFrame(json_data, ["id", "json_string"])
# تحليل سلاسل JSON إلى VariantType
variant_df = df.select(
"id",
parse_json("json_string").alias("data")
)
# الكتابة إلى ClickHouse باستخدام نوع JSON
variant_df.writeTo("clickhouse.default.user_data").create()
# أو حدِّد Variant بعدة أنواع
spark.sql("""
CREATE TABLE clickhouse.default.mixed_data (
id INT,
data VARIANT
) USING clickhouse
TBLPROPERTIES (
'clickhouse.column.data.variant_types' = 'String, Int64, Bool, JSON',
'engine' = 'MergeTree()',
'order_by' = 'id'
)
""")
```
```java theme={null}
import static org.apache.spark.sql.functions.*;
// إنشاء DataFrame يحتوي على بيانات JSON
List jsonData = Arrays.asList(
RowFactory.create(1, "{\"name\": \"Alice\", \"age\": 30}"),
RowFactory.create(2, "{\"name\": \"Bob\", \"age\": 25}"),
RowFactory.create(3, "{\"name\": \"Charlie\", \"city\": \"NYC\"}")
);
StructType schema = new StructType(new StructField[]{
DataTypes.createStructField("id", DataTypes.IntegerType, false),
DataTypes.createStructField("json_string", DataTypes.StringType, false)
});
Dataset jsonDF = spark.createDataFrame(jsonData, schema);
// تحليل سلاسل JSON إلى VariantType
Dataset variantDF = jsonDF.select(
col("id"),
parse_json(col("json_string")).as("data")
);
// الكتابة إلى ClickHouse باستخدام نوع JSON (كائنات JSON فقط)
variantDF.writeTo("clickhouse.default.user_data").create();
// أو حدِّد Variant بعدة أنواع
spark.sql("CREATE TABLE clickhouse.default.mixed_data (" +
"id INT, " +
"data VARIANT" +
") USING clickhouse " +
"TBLPROPERTIES (" +
"'clickhouse.column.data.variant_types' = 'String, Int64, Bool, JSON', " +
"'engine' = 'MergeTree()', " +
"'order_by' = 'id'" +
")");
```
### إنشاء جداول من نوع VariantType باستخدام Spark SQL
يمكنك إنشاء جداول من نوع VariantType باستخدام عبارات DDL في Spark SQL:
```sql theme={null}
-- Create table with JSON type (default)
CREATE TABLE clickhouse.default.json_table (
id INT,
data VARIANT
) USING clickhouse
TBLPROPERTIES (
'engine' = 'MergeTree()',
'order_by' = 'id'
)
```
```sql theme={null}
-- Create table with Variant type supporting multiple types
CREATE TABLE clickhouse.default.flexible_data (
id INT,
data VARIANT
) USING clickhouse
TBLPROPERTIES (
'clickhouse.column.data.variant_types' = 'String, Int64, Float64, Bool, Array(String), JSON',
'engine' = 'MergeTree()',
'order_by' = 'id'
)
```
### إعداد أنواع Variant
عند إنشاء جداول تضم أعمدة من نوع VariantType، يمكنك تحديد أنواع ClickHouse التي تريد استخدامها:
#### نوع JSON (الافتراضي)
إذا لم تُحدَّد الخاصية `variant_types`، فسيستخدم العمود نوع `JSON` الافتراضي في ClickHouse، والذي لا يقبل إلا كائنات JSON:
```sql theme={null}
CREATE TABLE clickhouse.default.json_table (
id INT,
data VARIANT
) USING clickhouse
TBLPROPERTIES (
'engine' = 'MergeTree()',
'order_by' = 'id'
)
```
ينشئ هذا استعلام ClickHouse التالي:
```sql theme={null}
CREATE TABLE json_table (id Int32, data JSON) ENGINE = MergeTree() ORDER BY id
```
#### نوع Variant مع أنواع متعددة
لدعم الأنواع الأولية والمصفوفات وكائنات JSON، حدِّد الأنواع في الخاصية `variant_types`:
```sql theme={null}
CREATE TABLE clickhouse.default.flexible_data (
id INT,
data VARIANT
) USING clickhouse
TBLPROPERTIES (
'clickhouse.column.data.variant_types' = 'String, Int64, Float64, Bool, Array(String), JSON',
'engine' = 'MergeTree()',
'order_by' = 'id'
)
```
ينشئ هذا استعلام ClickHouse التالي:
```sql theme={null}
CREATE TABLE flexible_data (
id Int32,
data Variant(String, Int64, Float64, Bool, Array(String), JSON)
) ENGINE = MergeTree() ORDER BY id
```
### أنواع Variant المدعومة
يمكن استخدام أنواع ClickHouse التالية في `Variant()`:
* **الأنواع الأساسية**: `String` و`Int8` و`Int16` و`Int32` و`Int64` و`UInt8` و`UInt16` و`UInt32` و`UInt64` و`Float32` و`Float64` و`Bool`
* **المصفوفات**: `Array(T)`، حيث يكون T أي نوع مدعوم، بما في ذلك المصفوفات المتداخلة
* **JSON**: `JSON` لتخزين كائنات JSON
### إعداد تنسيق القراءة
بشكل افتراضي، تُقرأ أعمدة JSON وVariant على أنها `VariantType`. يمكنك تجاوز هذا السلوك وقراءتها كسلاسل نصية:
```scala theme={null}
// قراءة JSON/Variant كسلاسل نصية بدلًا من VariantType
spark.conf.set("spark.clickhouse.read.jsonAs", "string")
val df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
// سيكون العمود data من النوع StringType ويحتوي على سلاسل JSON
```
```python theme={null}
# قراءة JSON/Variant كسلاسل نصية بدلًا من VariantType
spark.conf.set("spark.clickhouse.read.jsonAs", "string")
df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
# سيكون العمود data من النوع StringType ويحتوي على سلاسل JSON
```
```java theme={null}
// قراءة JSON/Variant كسلاسل نصية بدلًا من VariantType
spark.conf().set("spark.clickhouse.read.jsonAs", "string");
Dataset df = spark.sql("SELECT id, data FROM clickhouse.default.json_table");
// سيكون العمود data من النوع StringType ويحتوي على سلاسل JSON
```
### دعم صيغ الكتابة
يختلف دعم الكتابة لـ VariantType باختلاف الصيغة:
| التنسيق | الدعم | ملاحظات |
| ------- | ------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| JSON | ✅ كامل | يدعم النوعين `JSON` و`Variant` كليهما. يُوصى به لبيانات VariantType |
| Arrow | ⚠️ جزئي | يدعم الكتابة إلى النوع `JSON` في ClickHouse، لكنه لا يدعم النوع `Variant` في ClickHouse. الدعم الكامل بانتظار حل المشكلة [https://github.com/ClickHouse/ClickHouse/issues/92752](https://github.com/ClickHouse/ClickHouse/issues/92752) |
اضبط صيغة الكتابة:
```scala theme={null}
spark.conf.set("spark.clickhouse.write.format", "json") // Recommended for Variant types
```
إذا احتجت إلى الكتابة إلى نوع `Variant` في ClickHouse، فاستخدم تنسيق JSON. أما تنسيق Arrow فلا يدعم إلا الكتابة إلى النوع `JSON`.
### أفضل الممارسات
1. **استخدم نوع JSON للبيانات التي تحتوي على JSON فقط**: إذا كنت تخزّن كائنات JSON فقط، فاستخدم نوع JSON الافتراضي (من دون الخاصية `variant_types`)
2. **حدّد الأنواع صراحةً**: عند استخدام `Variant()`، اذكر صراحةً جميع الأنواع التي تنوي تخزينها
3. **فعّل الميزات التجريبية**: تأكد من تفعيل `allow_experimental_json_type = 1` في ClickHouse
4. **استخدم تنسيق JSON لعمليات الكتابة**: يُوصى باستخدام تنسيق JSON لبيانات VariantType لتحقيق توافق أفضل
5. **ضع أنماط الاستعلام في الاعتبار**: تدعم أنواع JSON/Variant استعلامات مسار JSON في ClickHouse لتصفية أكثر كفاءة
6. **تلميحات الأعمدة لتحسين الأداء**: عند استخدام حقول JSON في ClickHouse، يؤدي إضافة تلميحات الأعمدة إلى تحسين أداء الاستعلامات. لا تتوفر حاليًا إمكانية إضافة تلميحات الأعمدة عبر Spark. راجع [GitHub issue #497](https://github.com/ClickHouse/spark-clickhouse-connector/issues/497) لمتابعة هذه الميزة.
### مثال: سير عمل متكامل
```scala theme={null}
import org.apache.spark.sql.functions._
// Enable experimental JSON type in ClickHouse
spark.sql("SET allow_experimental_json_type = 1")
// Create table with Variant column
spark.sql("""
CREATE TABLE clickhouse.default.events (
event_id BIGINT,
event_time TIMESTAMP,
event_data VARIANT
) USING clickhouse
TBLPROPERTIES (
'clickhouse.column.event_data.variant_types' = 'String, Int64, Bool, JSON',
'engine' = 'MergeTree()',
'order_by' = 'event_time'
)
""")
// Prepare data with mixed types
val events = Seq(
(1L, "2024-01-01 10:00:00", """{"action": "login", "user_id": 123}"""),
(2L, "2024-01-01 10:05:00", """{"action": "purchase", "amount": 99.99}"""),
(3L, "2024-01-01 10:10:00", """{"action": "logout", "duration": 600}""")
).toDF("event_id", "event_time", "json_data")
// Convert to VariantType and write
val variantEvents = events.select(
col("event_id"),
to_timestamp(col("event_time")).as("event_time"),
parse_json(col("json_data")).as("event_data")
)
variantEvents.writeTo("clickhouse.default.events").append()
// Read and query
val result = spark.sql("""
SELECT event_id, event_time, event_data
FROM clickhouse.default.events
WHERE event_time >= '2024-01-01'
ORDER BY event_time
""")
result.show(false)
```
```python theme={null}
from pyspark.sql.functions import parse_json, to_timestamp
# Enable experimental JSON type in ClickHouse
spark.sql("SET allow_experimental_json_type = 1")
# Create table with Variant column
spark.sql("""
CREATE TABLE clickhouse.default.events (
event_id BIGINT,
event_time TIMESTAMP,
event_data VARIANT
) USING clickhouse
TBLPROPERTIES (
'clickhouse.column.event_data.variant_types' = 'String, Int64, Bool, JSON',
'engine' = 'MergeTree()',
'order_by' = 'event_time'
)
""")
# Prepare data with mixed types
events = [
(1, "2024-01-01 10:00:00", '{"action": "login", "user_id": 123}'),
(2, "2024-01-01 10:05:00", '{"action": "purchase", "amount": 99.99}'),
(3, "2024-01-01 10:10:00", '{"action": "logout", "duration": 600}')
]
df = spark.createDataFrame(events, ["event_id", "event_time", "json_data"])
# Convert to VariantType and write
variant_events = df.select(
"event_id",
to_timestamp("event_time").alias("event_time"),
parse_json("json_data").alias("event_data")
)
variant_events.writeTo("clickhouse.default.events").append()
# Read and query
result = spark.sql("""
SELECT event_id, event_time, event_data
FROM clickhouse.default.events
WHERE event_time >= '2024-01-01'
ORDER BY event_time
""")
result.show(truncate=False)
```
```java theme={null}
import static org.apache.spark.sql.functions.*;
// Enable experimental JSON type in ClickHouse
spark.sql("SET allow_experimental_json_type = 1");
// Create table with Variant column
spark.sql("CREATE TABLE clickhouse.default.events (" +
"event_id BIGINT, " +
"event_time TIMESTAMP, " +
"event_data VARIANT" +
") USING clickhouse " +
"TBLPROPERTIES (" +
"'clickhouse.column.event_data.variant_types' = 'String, Int64, Bool, JSON', " +
"'engine' = 'MergeTree()', " +
"'order_by' = 'event_time'" +
")");
// Prepare data with mixed types
List events = Arrays.asList(
RowFactory.create(1L, "2024-01-01 10:00:00", "{\"action\": \"login\", \"user_id\": 123}"),
RowFactory.create(2L, "2024-01-01 10:05:00", "{\"action\": \"purchase\", \"amount\": 99.99}"),
RowFactory.create(3L, "2024-01-01 10:10:00", "{\"action\": \"logout\", \"duration\": 600}")
);
StructType eventSchema = new StructType(new StructField[]{
DataTypes.createStructField("event_id", DataTypes.LongType, false),
DataTypes.createStructField("event_time", DataTypes.StringType, false),
DataTypes.createStructField("json_data", DataTypes.StringType, false)
});
Dataset eventsDF = spark.createDataFrame(events, eventSchema);
// Convert to VariantType and write
Dataset variantEvents = eventsDF.select(
col("event_id"),
to_timestamp(col("event_time")).as("event_time"),
parse_json(col("json_data")).as("event_data")
);
variantEvents.writeTo("clickhouse.default.events").append();
// Read and query
Dataset result = spark.sql("SELECT event_id, event_time, event_data " +
"FROM clickhouse.default.events " +
"WHERE event_time >= '2024-01-01' " +
"ORDER BY event_time");
result.show(false);
```
## الإعدادات
فيما يلي الإعدادات القابلة للتعديل المتاحة في الموصّل.
**استخدام الإعدادات**: هذه خيارات إعداد على مستوى Spark وتنطبق على كلٍّ من واجهة برمجة تطبيقات Catalog وTableProvider API. ويمكن ضبطها بإحدى الطريقتين التاليتين:
1. **إعداد Spark العام** (ينطبق على جميع العمليات):
```python theme={null}
spark.conf.set("spark.clickhouse.write.batchSize", "20000")
spark.conf.set("spark.clickhouse.write.compression.codec", "lz4")
```
2. **تجاوز على مستوى العملية** (في TableProvider API فقط — يمكنه تجاوز الإعدادات العامة):
```python theme={null}
df.write \
.format("clickhouse") \
.option("host", "your-host") \
.option("database", "default") \
.option("table", "my_table") \
.option("spark.clickhouse.write.batchSize", "20000") \
.option("spark.clickhouse.write.compression.codec", "lz4") \
.mode("append") \
.save()
```
بدلاً من ذلك، يمكنك ضبطها في `spark-defaults.conf` أو عند إنشاء جلسة Spark.
| المفتاح | القيمة الافتراضية | الوصف | منذ الإصدار |
| ------------------------------------------------------------------------ | -------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------- |
| spark.clickhouse.ignoreUnsupportedTransform | true | يدعم ClickHouse استخدام تعبيرات معقدة كمفاتيح تجزئة أو قيم partition، مثل `cityHash64(col_1, col_2)`، لكن Spark لا يدعمها حاليًا. إذا كانت القيمة `true`، فتجاهل التعبيرات غير المدعومة وسجّل تحذيرًا، وإلا فأخفِق فورًا مع استثناء. **تحذير**: عند ضبط `spark.clickhouse.write.distributed.convertLocal=true`، قد يؤدي تجاهل مفاتيح التجزئة غير المدعومة إلى تلف البيانات. يتحقق الموصّل من ذلك ويُصدر خطأً افتراضيًا. وللسماح بذلك، اضبط `spark.clickhouse.write.distributed.convertLocal.allowUnsupportedSharding=true` صراحةً. | 0.4.0 |
| spark.clickhouse.read.compression.codec | lz4 | خوارزمية الضغط المستخدمة لفك ضغط البيانات عند القراءة. خوارزميات الضغط المدعومة: none وlz4. | 0.5.0 |
| spark.clickhouse.read.distributed.convertLocal | true | عند القراءة من جدول Distributed، يُقرأ الجدول المحلي بدلًا منه. إذا كانت القيمة `true`، يتم تجاهل `spark.clickhouse.read.distributed.useClusterNodes`. | 0.1.0 |
| spark.clickhouse.read.fixedStringAs | binary | اقرأ نوع FixedString في ClickHouse كنوع بيانات Spark المحدد. الأنواع المدعومة: binary, string | 0.8.0 |
| spark.clickhouse.read.format | json | تنسيق التسلسل المستخدم للقراءة. التنسيقات المدعومة: json وbinary | 0.6.0 |
| spark.clickhouse.read.runtimeFilter.enabled | false | تمكين مرشح وقت التشغيل لعمليات القراءة. | 0.8.0 |
| spark.clickhouse.read.splitByPartitionId | true | إذا كانت القيمة `true`، فأنشئ عامل تصفية قسم الإدخال باستخدام العمود الافتراضي `_partition_id` بدلًا من قيمة القسم. توجد مشكلات معروفة في بناء شروط SQL استنادًا إلى قيمة القسم. تتطلب هذه الميزة ClickHouse Server v21.6+ | 0.4.0 |
| spark.clickhouse.useNullableQuerySchema | false | إذا كانت القيمة `true`، فاعتبر جميع حقول مخطط الاستعلام قابلة لـ NULL عند تنفيذ `CREATE/REPLACE TABLE ... AS SELECT ...` أثناء إنشاء الجدول. لاحظ أن هذا الإعداد يتطلب SPARK-43390 (المتوفر في Spark 3.5)، وبدون هذا التصحيح، فإنه يتصرف دائمًا كما لو كانت قيمته `true`. | 0.8.0 |
| spark.clickhouse.write.batchSize | 10000 | عدد السجلات لكل دفعة عند الكتابة إلى ClickHouse. | 0.1.0 |
| spark.clickhouse.write.compression.codec | lz4 | خوارزمية الضغط المستخدمة عند كتابة البيانات. خوارزميات الضغط المدعومة: none وlz4. | 0.3.0 |
| spark.clickhouse.write.distributed.convertLocal | false | عند الكتابة إلى جدول Distributed، تتم الكتابة إلى الجدول المحلي بدلًا منه. إذا كانت القيمة `true`، يتم تجاهل `spark.clickhouse.write.distributed.useClusterNodes`. يتجاوز هذا آلية التوجيه الأصلية في ClickHouse، ما يتطلب من Spark تقييم مفتاح التقسيم. عند استخدام تعبيرات تقسيم غير مدعومة، اضبط `spark.clickhouse.ignoreUnsupportedTransform` على `false` لمنع حدوث أخطاء صامتة في توزيع البيانات. | 0.1.0 |
| spark.clickhouse.write.distributed.convertLocal.allowUnsupportedSharding | false | اسمح بالكتابة إلى جداول `Distributed` مع `convertLocal=true` و`ignoreUnsupportedTransform=true` عندما يكون مفتاح التشظية غير مدعوم. هذا خيار خطير وقد يؤدي إلى تلف البيانات بسبب التشظية غير الصحيحة. عند ضبطه على `true`، يجب التأكد من أن بياناتك مرتبة/مُشظّاة بشكل صحيح قبل الكتابة، لأن Spark لا يمكنه تقييم تعبير التشظية غير المدعوم. لا تضبطه على `true` إلا إذا كنت تدرك المخاطر وقد تحققت من توزيع بياناتك. افتراضيًا، سيؤدي هذا الجمع إلى ظهور خطأ لمنع تلف البيانات الصامت. | 0.10.0 |
| spark.clickhouse.write.distributed.useClusterNodes | true | الكتابة إلى كل عُقد العنقود عند الكتابة إلى جدول Distributed. | 0.1.0 |
| spark.clickhouse.write.format | arrow | صيغة التسلسل المستخدمة للكتابة. التنسيقات المدعومة: json وarrow | 0.4.0 |
| spark.clickhouse.write.localSortByKey | true | إذا كانت القيمة `true`، فسيُجرى فرز محلي وفقًا لمفاتيح الفرز قبل الكتابة. | 0.3.0 |
| spark.clickhouse.write.localSortByPartition | القيمة الخاصة بـ spark.clickhouse.write.repartitionByPartition | إذا كانت القيمة `true`، فأجرِ فرزًا محليًا حسب التقسيم قبل الكتابة. وإذا لم تُعيَّن، فستكون مساويةً لـ `spark.clickhouse.write.repartitionByPartition`. | 0.3.0 |
| spark.clickhouse.write.maxRetry | 3 | الحد الأقصى لعدد مرات إعادة محاولة عملية كتابة دفعة واحدة إذا فشلت بسبب رموز خطأ قابلة لإعادة المحاولة. | 0.1.0 |
| spark.clickhouse.write.repartitionByPartition | true | ما إذا كان ينبغي إعادة تقسيم البيانات بحسب مفاتيح تقسيم ClickHouse لتتوافق مع توزيعات جدول ClickHouse قبل الكتابة. | 0.3.0 |
| spark.clickhouse.write.repartitionNum | 0 | يلزم إعادة تقسيم البيانات بما يتوافق مع توزيع جدول ClickHouse قبل الكتابة. استخدم هذا الإعداد لتحديد عدد أقسام إعادة التقسيم؛ وأي قيمة أقل من 1 تعني عدم الحاجة إلى ذلك. | 0.1.0 |
| spark.clickhouse.write.repartitionStrictly | false | إذا كانت `true`، فسيُوزّع Spark السجلات الواردة بدقة عبر الأقسام لتلبية التوزيع المطلوب قبل تمريرها إلى جدول مصدر البيانات عند الكتابة. وإلا، فقد يطبّق Spark بعض التحسينات لتسريع الاستعلام، لكن على حساب متطلب التوزيع. ملاحظة: يتطلب هذا الإعداد SPARK-37523 (المتوفرة في Spark 3.4)، وبدون هذا التصحيح سيعمل دائمًا كما لو كانت قيمته `true`. | 0.3.0 |
| spark.clickhouse.write.retryInterval | 10s | الفاصل الزمني، بالثواني، بين محاولات إعادة الكتابة. | 0.1.0 |
| spark.clickhouse.write.retryableErrorCodes | 241 | رموز الخطأ القابلة لإعادة المحاولة التي يعيدها خادم ClickHouse عند فشل عملية الكتابة. | 0.1.0 |
## أنواع البيانات المدعومة
يوضح هذا القسم كيفية توافق أنواع البيانات بين Spark وClickHouse. وتوفر الجداول أدناه مراجع سريعة
لتحويل أنواع البيانات عند قراءتها من ClickHouse إلى Spark، وعند إدراجها من Spark في ClickHouse.
### قراءة البيانات من ClickHouse إلى Spark
| نوع بيانات ClickHouse | نوع بيانات Spark | مدعوم | بدائي | ملاحظات |
| ----------------------------------------------------------------- | ------------------------------ | ----- | ----- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `Nothing` | `NullType` | ✅ | نعم | |
| `Bool` | `BooleanType` | ✅ | نعم | |
| `UInt8`, `Int16` | `ShortType` | ✅ | نعم | |
| `Int8` | `ByteType` | ✅ | نعم | |
| `UInt16`,`Int32` | `IntegerType` | ✅ | نعم | |
| `UInt32`,`Int64`, `UInt64` | `LongType` | ✅ | نعم | |
| `Int128`,`UInt128`, `Int256`, `UInt256` | `DecimalType(38, 0)` | ✅ | نعم | |
| `Float32` | `FloatType` | ✅ | نعم | |
| `Float64` | `DoubleType` | ✅ | نعم | |
| `String`, `UUID`, `Enum8`, `Enum16`, `IPv4`, `IPv6` | `StringType` | ✅ | نعم | |
| `FixedString` | `BinaryType`, `StringType` | ✅ | نعم | يتحكم فيه الإعداد `READ_FIXED_STRING_AS` |
| `Decimal` | `DecimalType` | ✅ | نعم | حتى دقة ومقياس `Decimal128` |
| `Decimal32` | `DecimalType(9, scale)` | ✅ | نعم | |
| `Decimal64` | `DecimalType(18, scale)` | ✅ | نعم | |
| `Decimal128` | `DecimalType(38, scale)` | ✅ | نعم | |
| `Date`, `Date32` | `DateType` | ✅ | نعم | |
| `DateTime`, `DateTime32`, `DateTime64` | `TimestampType` | ✅ | نعم | |
| `Array` | `ArrayType` | ✅ | لا | يُحوَّل أيضًا نوع عنصر الـArray |
| `Map` | `MapType` | ✅ | لا | المفاتيح مقتصرة على `StringType` |
| `IntervalYear` | `YearMonthIntervalType(Year)` | ✅ | نعم | |
| `IntervalMonth` | `YearMonthIntervalType(Month)` | ✅ | نعم | |
| `IntervalDay`, `IntervalHour`, `IntervalMinute`, `IntervalSecond` | `DayTimeIntervalType` | ✅ | لا | يُستخدم نوع interval المحدد |
| `JSON`, `Variant` | `VariantType` | ✅ | لا | يتطلب Spark 4.0+ وClickHouse 25.3+. ويمكن قراءته كـ `StringType` باستخدام `spark.clickhouse.read.jsonAs=string` |
| `Object` | | ❌ | | |
| `Nested` | | ❌ | | |
| `Tuple` | `StructType` | ✅ | لا | يدعم كلاً من `Tuple` المسمّاة وغير المسمّاة. تُطابَق الـ`Tuple` المسمّاة مع حقول `struct` حسب الاسم، بينما تستخدم الـ`Tuple` غير المسمّاة `_1` و`_2` وما إلى ذلك. كما يدعم `structs` المتداخلة والحقول القابلة للقيمة `NULL` |
| `Point` | | ❌ | | |
| `Polygon` | | ❌ | | |
| `MultiPolygon` | | ❌ | | |
| `Ring` | | ❌ | | |
| `IntervalQuarter` | | ❌ | | |
| `IntervalWeek` | | ❌ | | |
| `Decimal256` | | ❌ | | |
| `AggregateFunction` | | ❌ | | |
| `SimpleAggregateFunction` | | ❌ | | |
### إدراج البيانات من Spark إلى ClickHouse
| نوع بيانات Spark | نوع بيانات ClickHouse | مدعوم | بدائي | ملاحظات |
| ----------------------------------- | --------------------- | ----- | ----- | -------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `BooleanType` | `Bool` | ✅ | نعم | يُعيَّن إلى النوع `Bool` (وليس `UInt8`) بدءًا من الإصدار 0.9.0 |
| `ByteType` | `Int8` | ✅ | نعم | |
| `ShortType` | `Int16` | ✅ | نعم | |
| `IntegerType` | `Int32` | ✅ | نعم | |
| `LongType` | `Int64` | ✅ | نعم | |
| `FloatType` | `Float32` | ✅ | نعم | |
| `DoubleType` | `Float64` | ✅ | نعم | |
| `StringType` | `String` | ✅ | نعم | |
| `VarcharType` | `String` | ✅ | نعم | |
| `CharType` | `String` | ✅ | نعم | |
| `DecimalType` | `Decimal(p, s)` | ✅ | نعم | الدقة والمقياس حتى `Decimal128` |
| `DateType` | `Date` | ✅ | نعم | |
| `TimestampType` | `DateTime` | ✅ | نعم | |
| `ArrayType` (list, tuple, or array) | `Array` | ✅ | لا | يُحوَّل أيضًا نوع عنصر المصفوفة |
| `MapType` | `Map` | ✅ | لا | المفاتيح مقتصرة على `StringType` |
| `StructType` | `Tuple` | ✅ | لا | يُحوَّل إلى `Tuple` مُسمّى باستخدام أسماء الحقول. |
| `VariantType` | `JSON` or `Variant` | ✅ | لا | يتطلب Spark 4.0+ وClickHouse 25.3+. النوع الافتراضي هو `JSON`. استخدم الخاصية `clickhouse.column..variant_types` لتحديد `Variant` مع أنواع متعددة. |
| `Object` | | ❌ | | |
| `Nested` | | ❌ | | |
## المساهمة والدعم
إذا كنت ترغب في المساهمة في المشروع أو الإبلاغ عن أي مشكلات، فنحن نرحب بمشاركتك!
زُر [مستودع GitHub الخاص بنا](https://github.com/ClickHouse/spark-clickhouse-connector) لفتح مشكلة، أو اقتراح
تحسينات، أو إرسال طلب سحب.
نرحب بالمساهمات! يُرجى مراجعة إرشادات المساهمة في المستودع قبل البدء.
شكرًا لمساعدتك في تحسين ClickHouse Spark connector الخاص بنا!