`:
```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 API не поддерживает операции перезаписи на уровне партиций (например, режим `overwrite` с `partitionBy`). Работа над этой возможностью продолжается. Для отслеживания статуса см. [issue #34 на GitHub](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();
// Определение схемы для 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
// Определение схемы для 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
# Можно использовать любую другую комбинацию пакетов, совместимую согласно матрице совместимости выше.
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 — это промежуточный df 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 за один раз можно выполнить только один оператор.
```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, которые можно выполнять в приложении с помощью любого API — Java, Scala,
PySpark или оболочки shell.
## Работа с VariantType
Поддержка VariantType доступна в Spark 4.0+ и требует ClickHouse 25.3+ с включенными экспериментальными типами JSON/Variant.
Коннектор поддерживает тип `VariantType` в Spark для работы с полуструктурированными данными. VariantType сопоставляется с типами ClickHouse `JSON` и `Variant`, что позволяет эффективно хранить данные с гибкой схемой и выполнять по ним запросы.
Этот раздел посвящен исключительно сопоставлению и использованию 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}
-- Создать таблицу с типом JSON (по умолчанию)
CREATE TABLE clickhouse.default.json_table (
id INT,
data VARIANT
) USING clickhouse
TBLPROPERTIES (
'engine' = 'MergeTree()',
'order_by' = 'id'
)
```
```sql theme={null}
-- Создание таблицы с типом Variant, поддерживающим несколько типов
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
В `Variant()` можно использовать следующие типы ClickHouse:
* **Примитивные типы**: `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 зависит от формата:
| Format | Support | Notes |
| ------ | ------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| JSON | ✅ Полная | Поддерживает типы `JSON` и `Variant`. Рекомендуется для данных VariantType |
| Arrow | ⚠️ Частичная | Поддерживает запись в тип ClickHouse `JSON`. Не поддерживает тип ClickHouse `Variant`. Полная поддержка появится после решения [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") // Рекомендуется для типов Variant
```
Если вам нужно записывать данные в тип ClickHouse `Variant`, используйте формат JSON. Формат Arrow поддерживает запись только в тип `JSON`.
### Лучшие практики
1. **Используйте тип JSON для данных только в формате JSON**: Если вы храните только объекты JSON, используйте тип JSON по умолчанию (без свойства `variant_types`)
2. **Явно указывайте типы**: При использовании `Variant()` явно перечислите все типы, которые планируете хранить
3. **Включите экспериментальные возможности**: Убедитесь, что в ClickHouse включен параметр `allow_experimental_json_type = 1`
4. **Используйте формат JSON для записи**: Для данных VariantType рекомендуется формат JSON, так как он обеспечивает лучшую совместимость
5. **Учитывайте шаблоны запросов**: Типы JSON/Variant поддерживают в ClickHouse запросы по путям JSON для эффективной фильтрации
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._
// Включить экспериментальный тип JSON в ClickHouse
spark.sql("SET allow_experimental_json_type = 1")
// Создать таблицу со столбцом типа Variant
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'
)
""")
// Подготовить данные со смешанными типами
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")
// Преобразовать в VariantType и записать
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()
// Чтение и выполнение запроса
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
# Включить экспериментальный тип JSON в ClickHouse
spark.sql("SET allow_experimental_json_type = 1")
# Создать таблицу со столбцом Variant
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'
)
""")
# Подготовить данные со смешанными типами
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"])
# Преобразовать в VariantType и записать
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()
# Чтение и выполнение запроса
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.*;
// Включить экспериментальный тип JSON в ClickHouse
spark.sql("SET allow_experimental_json_type = 1");
// Создать таблицу со столбцом типа Variant
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'" +
")");
// Подготовить данные со смешанными типами
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);
// Преобразовать в VariantType и записать
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();
// Чтение и выполнение запроса
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 API, так и к 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 поддерживает использование сложных выражений в качестве ключей сегментирования или значений партиций, например `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 | Включить runtime filter для чтения. | 0.8.0 |
| spark.clickhouse.read.splitByPartitionId | true | Если `true`, формировать фильтр входных партиций по виртуальному столбцу `_partition_id`, а не по значению партиции. Есть известные проблемы со сборкой SQL-предикатов по значению партиции. Для этой возможности требуется ClickHouse Server v21.6+ | 0.4.0 |
| spark.clickhouse.useNullableQuerySchema | false | Если `true`, при выполнении `CREATE/REPLACE TABLE ... AS SELECT ...` все поля схемы запроса при создании таблицы помечаются как Nullable. Обратите внимание: для этой конфигурации требуется 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 таблицу записывайте данные в локальную таблицу вместо самой 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` | ✅ | Да | Точность и scale — до `Decimal128` |
| `Decimal32` | `DecimalType(9, scale)` | ✅ | Да | |
| `Decimal64` | `DecimalType(18, scale)` | ✅ | Да | |
| `Decimal128` | `DecimalType(38, scale)` | ✅ | Да | |
| `Date`, `Date32` | `DateType` | ✅ | Да | |
| `DateTime`, `DateTime32`, `DateTime64` | `TimestampType` | ✅ | Да | |
| `Array` | `ArrayType` | ✅ | Нет | Тип элементов массива также преобразуется |
| `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 по имени, а для неименованных используются `_1`, `_2` и т. д. Также поддерживаются вложенные struct и поля Nullable |
| `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)` | ✅ | Да | Точность и scale поддерживаются до `Decimal128` |
| `DateType` | `Date` | ✅ | Да | |
| `TimestampType` | `DateTime` | ✅ | Да | |
| `ArrayType` (list, tuple, or array) | `Array` | ✅ | Нет | Тип элементов массива также преобразуется |
| `MapType` | `Map` | ✅ | Нет | Ключи ограничены типом `StringType` |
| `StructType` | `Tuple` | ✅ | Нет | Преобразуется в именованный `Tuple` с именами полей. |
| `VariantType` | `JSON` или `Variant` | ✅ | Нет | Требует Spark 4.0+ и ClickHouse 25.3+. По умолчанию используется тип `JSON`. Используйте свойство `clickhouse.column..variant_types`, чтобы указать `Variant` с несколькими типами. |
| `Object` | | ❌ | | |
| `Nested` | | ❌ | | |
## Участие в проекте и поддержка
Если вы хотите внести вклад в проект или сообщить о проблеме, мы будем рады вашей помощи!
Посетите наш [репозиторий GitHub](https://github.com/ClickHouse/spark-clickhouse-connector), чтобы создать issue, предложить
улучшения или отправить pull request.
Мы приветствуем ваш вклад! Перед началом работы ознакомьтесь с рекомендациями по участию в репозитории.
Спасибо, что помогаете улучшать наш коннектор ClickHouse Spark!