` na configuração do Spark:
```text theme={null}
spark.sql.catalog.clickhouse.option.allow_nullable_key 1
spark.sql.catalog.clickhouse.option.index_granularity 8192
```
Ou defina essas configurações ao criar tabelas via 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'
)
```
## Configurações do ClickHouse Cloud
Ao se conectar ao [ClickHouse Cloud](https://clickhouse.com), certifique-se de habilitar o SSL e definir o modo de SSL adequado. Por exemplo:
```text theme={null}
spark.sql.catalog.clickhouse.option.ssl true
spark.sql.catalog.clickhouse.option.ssl_mode NONE
```
## Ler dados
```java theme={null}
public static void main(String[] args) {
// Criar uma sessão do 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;
```
## Gravar dados
**Sobrescrita de partição sem suporte**: Atualmente, a Catalog API não oferece suporte a operações de sobrescrita em nível de partição (por exemplo, o modo `overwrite` com `partitionBy`). Esse recurso está sendo desenvolvido. Consulte a [issue #34 no GitHub](https://github.com/ClickHouse/spark-clickhouse-connector/issues/34) para acompanhar o andamento desse recurso.
```java theme={null}
public static void main(String[] args) throws AnalysisException {
// Criar uma sessão do 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();
// Definir o esquema do 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")
);
// Criar um DataFrame
Dataset df = spark.createDataFrame(data, schema);
df.writeTo("clickhouse.default.example_table").append();
spark.stop();
}
```
```java theme={null}
object NativeSparkWrite extends App {
// Criar uma sessão do 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
// Definir o esquema do DataFrame
val rows = Seq(Row(1, "John"), Row(2, "Doe"))
val schema = List(
StructField("id", DataTypes.IntegerType, nullable = false),
StructField("name", StringType, nullable = true)
)
// Criar o 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
# Você pode usar qualquer outra combinação de pacotes que atenda à Matriz de Compatibilidade fornecida acima.
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")
# Criar o DataFrame
data = [Row(id=11, name="John"), Row(id=12, name="Doe")]
df = spark.createDataFrame(data)
# Gravar o DataFrame no ClickHouse
df.writeTo("clickhouse.default.example_table").append()
```
```sql theme={null}
-- resultTable é o df intermediário do Spark que queremos inserir em clickhouse.default.example_table
INSERT INTO TABLE clickhouse.default.example_table
SELECT * FROM resultTable;
```
## Operações DDL
Você pode executar operações DDL na sua instância do ClickHouse usando o Spark SQL, com todas as alterações sendo persistidas imediatamente no
ClickHouse.
O Spark SQL permite escrever consultas exatamente como no ClickHouse,
para que você possa executar diretamente comandos como CREATE TABLE, TRUNCATE e outros, sem nenhuma modificação, por exemplo:
Ao usar o Spark SQL, apenas uma instrução pode ser executada por vez.
```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
);
```
Os exemplos acima demonstram consultas em Spark SQL, que você pode executar no seu aplicativo usando qualquer API — Java, Scala,
PySpark ou shell.
## Trabalhando com VariantType
O suporte a VariantType está disponível no Spark 4.0+ e requer o ClickHouse 25.3+ com os tipos JSON/Variant experimentais habilitados.
O conector oferece suporte ao `VariantType` do Spark para trabalhar com dados semiestruturados. O VariantType é mapeado para os tipos `JSON` e `Variant` do ClickHouse, permitindo armazenar e consultar com eficiência dados com esquema flexível.
Esta seção se concentra especificamente no mapeamento e uso do VariantType. Para uma visão geral completa de todos os tipos de dados compatíveis, consulte a seção [Tipos de dados compatíveis](#supported-data-types).
### Mapeamento de tipos do ClickHouse
| Tipo do ClickHouse | Tipo do Spark | Descrição |
| ---------------------- | ------------- | ---------------------------------------------------------------- |
| `JSON` | `VariantType` | Armazena apenas objetos JSON (deve começar com `{`) |
| `Variant(T1, T2, ...)` | `VariantType` | Armazena vários tipos, incluindo tipos primitivos, arrays e JSON |
### Lendo dados do VariantType
Ao ler do ClickHouse, as colunas `JSON` e `Variant` são mapeadas automaticamente para o `VariantType` do Spark:
```scala theme={null}
// Ler coluna JSON como VariantType
val df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
// Acessar dados Variant
df.show()
// Converter Variant em string JSON para inspeção
import org.apache.spark.sql.functions._
df.select(
col("id"),
to_json(col("data")).as("data_json")
).show()
```
```python theme={null}
# Ler coluna JSON como VariantType
df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
# Acessar dados Variant
df.show()
# Converter Variant em string JSON para inspeção
from pyspark.sql.functions import to_json
df.select(
"id",
to_json("data").alias("data_json")
).show()
```
```java theme={null}
// Ler coluna JSON como VariantType
Dataset df = spark.sql("SELECT id, data FROM clickhouse.default.json_table");
// Acessar dados Variant
df.show();
// Converter Variant em string JSON para inspeção
import static org.apache.spark.sql.functions.*;
df.select(
col("id"),
to_json(col("data")).as("data_json")
).show();
```
### Gravando dados VariantType
Você pode gravar dados VariantType no ClickHouse usando tipos de coluna JSON ou Variant:
```scala theme={null}
import org.apache.spark.sql.functions._
// Criar DataFrame com dados JSON
val jsonData = Seq(
(1, """{"name": "Alice", "age": 30}"""),
(2, """{"name": "Bob", "age": 25}"""),
(3, """{"name": "Charlie", "city": "NYC"}""")
).toDF("id", "json_string")
// Converter strings JSON em VariantType
val variantDF = jsonData.select(
col("id"),
parse_json(col("json_string")).as("data")
)
// Gravar no ClickHouse com o tipo JSON (somente objetos JSON)
variantDF.writeTo("clickhouse.default.user_data").create()
// Ou especificar Variant com vários tipos
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
# Criar DataFrame com dados 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"])
# Converter strings JSON em VariantType
variant_df = df.select(
"id",
parse_json("json_string").alias("data")
)
# Gravar no ClickHouse com o tipo JSON
variant_df.writeTo("clickhouse.default.user_data").create()
# Ou especificar Variant com vários tipos
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.*;
// Criar DataFrame com dados 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);
// Converter strings JSON em VariantType
Dataset variantDF = jsonDF.select(
col("id"),
parse_json(col("json_string")).as("data")
);
// Gravar no ClickHouse com o tipo JSON (somente objetos JSON)
variantDF.writeTo("clickhouse.default.user_data").create();
// Ou especificar Variant com vários tipos
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'" +
")");
```
### Criando tabelas do tipo VariantType com Spark SQL
Você pode criar tabelas do tipo VariantType usando DDL do Spark SQL:
```sql theme={null}
-- Criar tabela com tipo JSON (padrão)
CREATE TABLE clickhouse.default.json_table (
id INT,
data VARIANT
) USING clickhouse
TBLPROPERTIES (
'engine' = 'MergeTree()',
'order_by' = 'id'
)
```
```sql theme={null}
-- Criar tabela com tipo Variant suportando múltiplos tipos
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'
)
```
### Configurando os tipos Variant
Ao criar tabelas com colunas `VariantType`, você pode especificar quais tipos do ClickHouse devem ser usados:
#### Tipo JSON (padrão)
Se nenhuma propriedade `variant_types` for especificada, a coluna usará, por padrão, o tipo `JSON` do ClickHouse, que aceita apenas objetos JSON:
```sql theme={null}
CREATE TABLE clickhouse.default.json_table (
id INT,
data VARIANT
) USING clickhouse
TBLPROPERTIES (
'engine' = 'MergeTree()',
'order_by' = 'id'
)
```
Isso cria a seguinte consulta do ClickHouse:
```sql theme={null}
CREATE TABLE json_table (id Int32, data JSON) ENGINE = MergeTree() ORDER BY id
```
#### Tipo Variant com múltiplos tipos
Para dar suporte a primitivos, arrays e objetos JSON, especifique os tipos na propriedade `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'
)
```
Isso cria a seguinte consulta do ClickHouse:
```sql theme={null}
CREATE TABLE flexible_data (
id Int32,
data Variant(String, Int64, Float64, Bool, Array(String), JSON)
) ENGINE = MergeTree() ORDER BY id
```
### Tipos de Variant compatíveis
Os seguintes tipos do ClickHouse podem ser usados em `Variant()`:
* **Primitivos**: `String`, `Int8`, `Int16`, `Int32`, `Int64`, `UInt8`, `UInt16`, `UInt32`, `UInt64`, `Float32`, `Float64`, `Bool`
* **Arrays**: `Array(T)`, em que T é qualquer tipo compatível, incluindo arrays aninhados
* **JSON**: `JSON` para armazenar objetos JSON
### Configuração do formato de leitura
Por padrão, as colunas JSON e Variant são lidas como `VariantType`. Você pode alterar esse comportamento para lê-las como strings:
```scala theme={null}
// Lê JSON/Variant como strings em vez de VariantType
spark.conf.set("spark.clickhouse.read.jsonAs", "string")
val df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
// a coluna data será do tipo StringType e conterá strings JSON
```
```python theme={null}
# Lê JSON/Variant como strings em vez de VariantType
spark.conf.set("spark.clickhouse.read.jsonAs", "string")
df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
# a coluna data será do tipo StringType e conterá strings JSON
```
```java theme={null}
// Lê JSON/Variant como strings em vez de VariantType
spark.conf().set("spark.clickhouse.read.jsonAs", "string");
Dataset df = spark.sql("SELECT id, data FROM clickhouse.default.json_table");
// a coluna data será do tipo StringType e conterá strings JSON
```
### Suporte ao formato de gravação
O suporte à gravação para VariantType varia de acordo com o formato:
| Formato | Suporte | Observações |
| ------- | ---------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| JSON | ✅ Completo | Oferece suporte aos tipos `JSON` e `Variant`. Recomendado para dados VariantType |
| Arrow | ⚠️ Parcial | Oferece suporte à gravação no tipo `JSON` do ClickHouse. Não oferece suporte ao tipo `Variant` do ClickHouse. O suporte completo depende da resolução de [https://github.com/ClickHouse/ClickHouse/issues/92752](https://github.com/ClickHouse/ClickHouse/issues/92752) |
Configure o formato de gravação:
```scala theme={null}
spark.conf.set("spark.clickhouse.write.format", "json") // Recomendado para tipos Variant
```
Se você precisar gravar em um tipo `Variant` do ClickHouse, use o formato JSON. O formato Arrow só oferece suporte à gravação apenas no tipo `JSON`.
### Melhores práticas
1. **Use o tipo JSON para dados exclusivamente em JSON**: Se você armazena apenas objetos JSON, use o tipo JSON padrão (sem a propriedade `variant_types`)
2. **Especifique os tipos explicitamente**: Ao usar `Variant()`, liste explicitamente todos os tipos que você pretende armazenar
3. **Habilite recursos experimentais**: Verifique se o ClickHouse está com `allow_experimental_json_type = 1` habilitado
4. **Use o formato JSON para escritas**: O formato JSON é recomendado para dados do VariantType, por oferecer melhor compatibilidade
5. **Considere os padrões de consulta**: Os tipos JSON/Variant oferecem suporte às consultas de caminho JSON do ClickHouse para uma filtragem eficiente
6. **Column hints para desempenho**: Ao usar campos JSON no ClickHouse, adicionar column hints melhora o desempenho da consulta. No momento, não há suporte para adicionar column hints via Spark. Consulte a [GitHub issue #497](https://github.com/ClickHouse/spark-clickhouse-connector/issues/497) para acompanhar esse recurso.
### Exemplo: Fluxo de trabalho completo
```scala theme={null}
import org.apache.spark.sql.functions._
// Habilitar o tipo JSON experimental no ClickHouse
spark.sql("SET allow_experimental_json_type = 1")
// Criar tabela com coluna 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'
)
""")
// Preparar dados com tipos mistos
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")
// Converter para VariantType e gravar
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()
// Ler e consultar
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
# Habilitar o tipo JSON experimental no ClickHouse
spark.sql("SET allow_experimental_json_type = 1")
# Criar tabela com coluna 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'
)
""")
# Preparar dados com tipos mistos
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"])
# Converter para VariantType e gravar
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()
# Ler e consultar
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.*;
// Habilitar o tipo JSON experimental no ClickHouse
spark.sql("SET allow_experimental_json_type = 1");
// Criar tabela com coluna 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'" +
")");
// Preparar dados com tipos mistos
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);
// Converter para VariantType e gravar
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();
// Ler e consultar
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);
```
## Configurações
A seguir estão as configurações ajustáveis disponíveis no conector.
**Uso das configurações**: estas são opções de configuração no nível do Spark que se aplicam tanto à Catalog API quanto à TableProvider API. Elas podem ser definidas de duas formas:
1. **Configuração global do Spark** (aplica-se a todas as operações):
```python theme={null}
spark.conf.set("spark.clickhouse.write.batchSize", "20000")
spark.conf.set("spark.clickhouse.write.compression.codec", "lz4")
```
2. **Substituição por operação** (somente na TableProvider API — pode sobrescrever as configurações globais):
```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()
```
Como alternativa, defina essas opções em `spark-defaults.conf` ou ao criar a sessão do Spark.
| Chave | Padrão | Descrição | Desde |
| ------------------------------------------------------------------------ | ------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------ |
| spark.clickhouse.ignoreUnsupportedTransform | true | O ClickHouse oferece suporte ao uso de expressões complexas como chaves de sharding ou valores de partição, por exemplo, `cityHash64(col_1, col_2)`, mas atualmente o Spark não consegue oferecer suporte a isso. Se `true`, ignora as expressões sem suporte e registra um aviso; caso contrário, falha imediatamente com uma exceção. **Aviso**: quando `spark.clickhouse.write.distributed.convertLocal=true`, ignorar chaves de sharding sem suporte pode corromper os dados. O connector valida isso e gera um erro por padrão. Para permitir esse comportamento, defina explicitamente `spark.clickhouse.write.distributed.convertLocal.allowUnsupportedSharding=true`. | 0.4.0 |
| spark.clickhouse.read.compression.codec | lz4 | O codec usado para descomprimir os dados para leitura. Codecs compatíveis: none, lz4. | 0.5.0 |
| spark.clickhouse.read.distributed.convertLocal | true | Ao ler uma tabela Distributed, leia a tabela local em vez da própria tabela Distributed. Se `true`, ignore `spark.clickhouse.read.distributed.useClusterNodes`. | 0.1.0 |
| spark.clickhouse.read.fixedStringAs | binary | Lê o tipo FixedString do ClickHouse como o tipo de dados do Spark especificado. Tipos compatíveis: binary, string | 0.8.0 |
| spark.clickhouse.read.format | json | Formato de serialização para leitura. Formatos suportados: json, binary | 0.6.0 |
| spark.clickhouse.read.runtimeFilter.enabled | false | Habilita o filtro em tempo de execução para leitura. | 0.8.0 |
| spark.clickhouse.read.splitByPartitionId | true | Se `true`, constrói o filtro da partição de entrada pela coluna virtual `_partition_id`, em vez do valor da partição. Há problemas conhecidos na construção de predicados SQL com base no valor da partição. Este recurso requer o ClickHouse Server v21.6+ | 0.4.0 |
| spark.clickhouse.useNullableQuerySchema | false | Se `true`, marque todos os campos do esquema da consulta como anuláveis ao criar a tabela com `CREATE/REPLACE TABLE ... AS SELECT ...`. Observe que essa configuração requer o SPARK-43390 (disponível no Spark 3.5); sem esse patch, ela sempre se comporta como `true`. | 0.8.0 |
| spark.clickhouse.write.batchSize | 10000 | O número de registros por Batch ao gravar no ClickHouse. | 0.1.0 |
| spark.clickhouse.write.compression.codec | lz4 | O codec usado para compactar os dados durante a gravação. Codecs compatíveis: none, lz4. | 0.3.0 |
| spark.clickhouse.write.distributed.convertLocal | false | Ao gravar em uma tabela Distributed, grave na tabela local em vez da própria tabela Distributed. Se estiver como `true`, ignore `spark.clickhouse.write.distributed.useClusterNodes`. Isso contorna o roteamento nativo do ClickHouse, fazendo com que o Spark precise avaliar a chave de sharding. Ao usar expressões de sharding não suportadas, defina `spark.clickhouse.ignoreUnsupportedTransform` como `false` para evitar erros silenciosos na distribuição de dados. | 0.1.0 |
| spark.clickhouse.write.distributed.convertLocal.allowUnsupportedSharding | false | Permite gravar em tabelas Distributed com `convertLocal=true` e `ignoreUnsupportedTransform=true` quando a chave de sharding não é suportada. Isso é perigoso e pode causar corrupção de dados devido a sharding incorreto. Quando definido como `true`, você deve garantir que seus dados estejam devidamente ordenados/distribuídos entre shards antes da gravação, pois o Spark não consegue avaliar a expressão de sharding não suportada. Defina como `true` somente se você compreender os riscos e tiver verificado a distribuição dos seus dados. Por padrão, essa combinação gerará um erro para evitar corrupção silenciosa de dados. | 0.10.0 |
| spark.clickhouse.write.distributed.useClusterNodes | true | Grava em todos os nós do cluster ao gravar em uma tabela Distributed. | 0.1.0 |
| spark.clickhouse.write.format | arrow | Formato de serialização para gravação. Formatos compatíveis: json, arrow | 0.4.0 |
| spark.clickhouse.write.localSortByKey | true | Se `true`, realiza a ordenação local pelas chaves de ordenação antes da gravação. | 0.3.0 |
| spark.clickhouse.write.localSortByPartition | valor de spark.clickhouse.write.repartitionByPartition | Se `true`, faz a ordenação local por partição antes da gravação. Se não for definido, é igual a `spark.clickhouse.write.repartitionByPartition`. | 0.3.0 |
| spark.clickhouse.write.maxRetry | 3 | O número máximo de tentativas de gravação para uma única gravação em lote que falhou com códigos passíveis de retry. | 0.1.0 |
| spark.clickhouse.write.repartitionByPartition | true | Se os dados devem ser reparticionados com base nas chaves de partição do ClickHouse para corresponder à distribuição da tabela do ClickHouse antes da gravação. | 0.3.0 |
| spark.clickhouse.write.repartitionNum | 0 | É necessário reparticionar os dados para que correspondam à distribuição da tabela do ClickHouse antes da gravação; use esta configuração para especificar o número de repartições; valores menores que 1 significam que não há necessidade de reparticionamento. | 0.1.0 |
| spark.clickhouse.write.repartitionStrictly | false | Se `true`, o Spark distribuirá rigorosamente os registros de entrada entre as partições para atender à distribuição exigida antes de gravar os registros na tabela da fonte de dados. Caso contrário, o Spark poderá aplicar certas otimizações para acelerar a consulta, mas violar o requisito de distribuição. Observe que esta configuração requer o SPARK-37523 (disponível no Spark 3.4); sem esse patch, ela sempre se comporta como `true`. | 0.3.0 |
| spark.clickhouse.write.retryInterval | 10s | O intervalo, em segundos, entre tentativas de gravação. | 0.1.0 |
| spark.clickhouse.write.retryableErrorCodes | 241 | Os códigos de erro retornados pelo servidor ClickHouse que permitem nova tentativa quando a gravação falha. | 0.1.0 |
## Tipos de dados compatíveis
Esta seção apresenta o mapeamento dos tipos de dados entre o Spark e o ClickHouse. As tabelas abaixo servem como referência rápida
para a conversão de tipos de dados ao ler dados do ClickHouse no Spark e ao inserir dados do Spark no ClickHouse.
### Lendo dados do ClickHouse no Spark
| Tipo de dados do ClickHouse | Tipo de dados do Spark | Suportado | É primitivo | Observações |
| ----------------------------------------------------------------- | ------------------------------ | --------- | ----------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `Nothing` | `NullType` | ✅ | Sim | |
| `Bool` | `BooleanType` | ✅ | Sim | |
| `UInt8`, `Int16` | `ShortType` | ✅ | Sim | |
| `Int8` | `ByteType` | ✅ | Sim | |
| `UInt16`,`Int32` | `IntegerType` | ✅ | Sim | |
| `UInt32`,`Int64`, `UInt64` | `LongType` | ✅ | Sim | |
| `Int128`,`UInt128`, `Int256`, `UInt256` | `DecimalType(38, 0)` | ✅ | Sim | |
| `Float32` | `FloatType` | ✅ | Sim | |
| `Float64` | `DoubleType` | ✅ | Sim | |
| `String`, `UUID`, `Enum8`, `Enum16`, `IPv4`, `IPv6` | `StringType` | ✅ | Sim | |
| `FixedString` | `BinaryType`, `StringType` | ✅ | Sim | Controlado pela configuração `READ_FIXED_STRING_AS` |
| `Decimal` | `DecimalType` | ✅ | Sim | Precisão e escala de até `Decimal128` |
| `Decimal32` | `DecimalType(9, scale)` | ✅ | Sim | |
| `Decimal64` | `DecimalType(18, scale)` | ✅ | Sim | |
| `Decimal128` | `DecimalType(38, scale)` | ✅ | Sim | |
| `Date`, `Date32` | `DateType` | ✅ | Sim | |
| `DateTime`, `DateTime32`, `DateTime64` | `TimestampType` | ✅ | Sim | |
| `Array` | `ArrayType` | ✅ | Não | O tipo de elemento do Array também é convertido |
| `Map` | `MapType` | ✅ | Não | As chaves são limitadas a `StringType` |
| `IntervalYear` | `YearMonthIntervalType(Year)` | ✅ | Sim | |
| `IntervalMonth` | `YearMonthIntervalType(Month)` | ✅ | Sim | |
| `IntervalDay`, `IntervalHour`, `IntervalMinute`, `IntervalSecond` | `DayTimeIntervalType` | ✅ | Não | O tipo de intervalo específico é usado |
| `JSON`, `Variant` | `VariantType` | ✅ | Não | Requer Spark 4.0+ e ClickHouse 25.3+. Pode ser lido como `StringType` com `spark.clickhouse.read.jsonAs=string` |
| `Object` | | ❌ | | |
| `Nested` | | ❌ | | |
| `Tuple` | `StructType` | ✅ | Não | Suporta Tuples nomeadas e não nomeadas. As Tuples nomeadas são mapeadas para campos da struct pelo nome; as não nomeadas usam `_1`, `_2` etc. Suporta structs aninhadas e campos Nullable |
| `Point` | | ❌ | | |
| `Polygon` | | ❌ | | |
| `MultiPolygon` | | ❌ | | |
| `Ring` | | ❌ | | |
| `IntervalQuarter` | | ❌ | | |
| `IntervalWeek` | | ❌ | | |
| `Decimal256` | | ❌ | | |
| `AggregateFunction` | | ❌ | | |
| `SimpleAggregateFunction` | | ❌ | | |
### Inserindo dados do Spark no ClickHouse
| Tipo de dado do Spark | Tipo de dado do ClickHouse | Suportado | É primitivo | Observações |
| ----------------------------------- | -------------------------- | --------- | ----------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `BooleanType` | `Bool` | ✅ | Sim | Mapeado para o tipo `Bool` (e não `UInt8`) desde a versão 0.9.0 |
| `ByteType` | `Int8` | ✅ | Sim | |
| `ShortType` | `Int16` | ✅ | Sim | |
| `IntegerType` | `Int32` | ✅ | Sim | |
| `LongType` | `Int64` | ✅ | Sim | |
| `FloatType` | `Float32` | ✅ | Sim | |
| `DoubleType` | `Float64` | ✅ | Sim | |
| `StringType` | `String` | ✅ | Sim | |
| `VarcharType` | `String` | ✅ | Sim | |
| `CharType` | `String` | ✅ | Sim | |
| `DecimalType` | `Decimal(p, s)` | ✅ | Sim | Precisão e scale até `Decimal128` |
| `DateType` | `Date` | ✅ | Sim | |
| `TimestampType` | `DateTime` | ✅ | Sim | |
| `ArrayType` (lista, tupla ou array) | `Array` | ✅ | Não | O element type do array também é convertido |
| `MapType` | `Map` | ✅ | Não | As chaves são limitadas a `StringType` |
| `StructType` | `Tuple` | ✅ | Não | Convertido em `Tuple` nomeada com nomes de field. |
| `VariantType` | `JSON` ou `Variant` | ✅ | Não | Requer Spark 4.0+ e ClickHouse 25.3+. O padrão é o tipo `JSON`. Use a propriedade `clickhouse.column..variant_types` para especificar `Variant` com vários tipos. |
| `Object` | | ❌ | | |
| `Nested` | | ❌ | | |
## Contribuições e suporte
Se você quiser contribuir com o projeto ou relatar algum problema, ficaremos felizes com sua colaboração!
Visite nosso [repositório no GitHub](https://github.com/ClickHouse/spark-clickhouse-connector) para abrir uma issue, sugerir
melhorias ou enviar um pull request.
Contribuições são bem-vindas! Consulte as diretrizes de contribuição no repositório antes de começar.
Agradecemos por ajudar a melhorar nosso ClickHouse Spark connector!