` dans votre configuration Spark :
```text theme={null}
spark.sql.catalog.clickhouse.option.allow_nullable_key 1
spark.sql.catalog.clickhouse.option.index_granularity 8192
```
Ou définissez-les au moment de créer des tables 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'
)
```
## Paramètres de ClickHouse Cloud
Lorsque vous vous connectez à [ClickHouse Cloud](https://clickhouse.com), veillez à activer SSL et à définir le mode SSL adéquat. Par exemple :
```text theme={null}
spark.sql.catalog.clickhouse.option.ssl true
spark.sql.catalog.clickhouse.option.ssl_mode NONE
```
## Lire les données
```java theme={null}
public static void main(String[] args) {
// Créer une session 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;
```
## Écrire des données
**L’écrasement de partitions n’est pas pris en charge** : l’API Catalog ne prend pas encore en charge les opérations d’écrasement au niveau des partitions (par ex. le mode `overwrite` avec `partitionBy`). Cette fonctionnalité est en cours de développement. Consultez l’[issue GitHub n°34](https://github.com/ClickHouse/spark-clickhouse-connector/issues/34) pour en suivre l’avancement.
```java theme={null}
public static void main(String[] args) throws AnalysisException {
// Créer une session 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();
// Définir le schéma du 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")
);
// Créer un DataFrame
Dataset df = spark.createDataFrame(data, schema);
df.writeTo("clickhouse.default.example_table").append();
spark.stop();
}
```
```java theme={null}
object NativeSparkWrite extends App {
// Créer une session 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
// Définir le schéma du DataFrame
val rows = Seq(Row(1, "John"), Row(2, "Doe"))
val schema = List(
StructField("id", DataTypes.IntegerType, nullable = false),
StructField("name", StringType, nullable = true)
)
// Créer le DataFrame
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
# N'hésitez pas à utiliser toute autre combinaison de packages respectant la matrice de compatibilité fournie ci-dessus.
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")
# Créer un DataFrame
data = [Row(id=11, name="John"), Row(id=12, name="Doe")]
df = spark.createDataFrame(data)
# Écrire le DataFrame dans ClickHouse
df.writeTo("clickhouse.default.example_table").append()
```
```sql theme={null}
-- resultTable est le DataFrame Spark intermédiaire que nous voulons insérer dans clickhouse.default.example_table
INSERT INTO TABLE clickhouse.default.example_table
SELECT * FROM resultTable;
```
## Opérations DDL
Vous pouvez effectuer des opérations DDL sur votre instance ClickHouse à l’aide de Spark SQL, toutes les modifications étant immédiatement enregistrées dans
ClickHouse.
Spark SQL vous permet d’écrire des requêtes exactement comme dans ClickHouse,
vous pouvez donc exécuter directement des commandes telles que CREATE TABLE, TRUNCATE, entre autres, sans aucune modification, par exemple :
Lors de l’utilisation de Spark SQL, une seule instruction peut être exécutée à la fois.
```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
);
```
Les exemples ci-dessus montrent des requêtes Spark SQL, que vous pouvez exécuter dans votre application à l’aide de n’importe quelle API — Java, Scala,
PySpark ou le shell.
## Travailler avec VariantType
La prise en charge de VariantType est disponible avec Spark 4.0+ et nécessite ClickHouse 25.3+ avec les types JSON/Variant expérimentaux activés.
Le connecteur prend en charge le `VariantType` de Spark pour manipuler des données semi-structurées. VariantType correspond aux types `JSON` et `Variant` de ClickHouse, ce qui vous permet de stocker et d'interroger efficacement des données à schéma souple.
Cette section porte spécifiquement sur la correspondance et l'utilisation de VariantType. Pour un aperçu complet de tous les types de données pris en charge, consultez la section [Types de données pris en charge](#supported-data-types).
### Correspondance des types ClickHouse
| Type ClickHouse | Type Spark | Description |
| ---------------------- | ------------- | ------------------------------------------------------------------------- |
| `JSON` | `VariantType` | Stocke uniquement des objets JSON (doit commencer par `{`) |
| `Variant(T1, T2, ...)` | `VariantType` | Stocke plusieurs types, notamment des primitives, des tableaux et du JSON |
### Lecture des données `VariantType`
Lors de la lecture depuis ClickHouse, les colonnes `JSON` et `Variant` sont automatiquement converties en `VariantType` de Spark :
```scala theme={null}
// Lire une colonne JSON en tant que VariantType
val df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
// Accéder aux données de type variant
df.show()
// Convertir le variant en chaîne JSON pour l'examiner
import org.apache.spark.sql.functions._
df.select(
col("id"),
to_json(col("data")).as("data_json")
).show()
```
```python theme={null}
# Lire une colonne JSON en tant que VariantType
df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
# Accéder aux données de type variant
df.show()
# Convertir le variant en chaîne JSON pour l'examiner
from pyspark.sql.functions import to_json
df.select(
"id",
to_json("data").alias("data_json")
).show()
```
```java theme={null}
// Lire une colonne JSON en tant que VariantType
Dataset df = spark.sql("SELECT id, data FROM clickhouse.default.json_table");
// Accéder aux données de type variant
df.show();
// Convertir le variant en chaîne JSON pour l'examiner
import static org.apache.spark.sql.functions.*;
df.select(
col("id"),
to_json(col("data")).as("data_json")
).show();
```
### Écrire des données de type VariantType
Vous pouvez écrire des données de type VariantType dans ClickHouse à l’aide de types de colonne JSON ou Variant :
```scala theme={null}
import org.apache.spark.sql.functions._
// Créer un DataFrame avec des données JSON
val jsonData = Seq(
(1, """{"name": "Alice", "age": 30}"""),
(2, """{"name": "Bob", "age": 25}"""),
(3, """{"name": "Charlie", "city": "NYC"}""")
).toDF("id", "json_string")
// Convertir les chaînes JSON en VariantType
val variantDF = jsonData.select(
col("id"),
parse_json(col("json_string")).as("data")
)
// Écrire dans ClickHouse avec le type JSON (objets JSON uniquement)
variantDF.writeTo("clickhouse.default.user_data").create()
// Ou définir Variant avec plusieurs types
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
# Créer un DataFrame avec des données 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"])
# Convertir les chaînes JSON en VariantType
variant_df = df.select(
"id",
parse_json("json_string").alias("data")
)
# Écrire dans ClickHouse avec le type JSON
variant_df.writeTo("clickhouse.default.user_data").create()
# Ou définir Variant avec plusieurs types
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.*;
// Créer un DataFrame avec des données 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);
// Convertir les chaînes JSON en VariantType
Dataset variantDF = jsonDF.select(
col("id"),
parse_json(col("json_string")).as("data")
);
// Écrire dans ClickHouse avec le type JSON (objets JSON uniquement)
variantDF.writeTo("clickhouse.default.user_data").create();
// Ou définir Variant avec plusieurs types
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'" +
")");
```
### Créer des tables VariantType avec Spark SQL
Vous pouvez créer des tables VariantType à l’aide du DDL de 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'
)
```
### Configuration des types de Variant
Lors de la création de tables comportant des colonnes `VariantType`, vous pouvez spécifier les types ClickHouse à utiliser :
#### Type JSON (par défaut)
Si aucune propriété `variant_types` n’est spécifiée, la colonne utilise par défaut le type `JSON` de ClickHouse, qui n’accepte que les objets JSON :
```sql theme={null}
CREATE TABLE clickhouse.default.json_table (
id INT,
data VARIANT
) USING clickhouse
TBLPROPERTIES (
'engine' = 'MergeTree()',
'order_by' = 'id'
)
```
Cela génère la requête ClickHouse suivante :
```sql theme={null}
CREATE TABLE json_table (id Int32, data JSON) ENGINE = MergeTree() ORDER BY id
```
#### Type Variant avec plusieurs types
Pour prendre en charge les types primitifs, les tableaux et les objets JSON, indiquez les types dans la propriété `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'
)
```
Cela génère la requête ClickHouse suivante :
```sql theme={null}
CREATE TABLE flexible_data (
id Int32,
data Variant(String, Int64, Float64, Bool, Array(String), JSON)
) ENGINE = MergeTree() ORDER BY id
```
### Types pris en charge dans Variant
Les types ClickHouse suivants peuvent être utilisés dans `Variant()` :
* **Primitives** : `String`, `Int8`, `Int16`, `Int32`, `Int64`, `UInt8`, `UInt16`, `UInt32`, `UInt64`, `Float32`, `Float64`, `Bool`
* **Tableaux** : `Array(T)`, où T est n’importe quel type pris en charge, y compris des tableaux imbriqués
* **JSON** : `JSON` pour stocker des objets JSON
### Configuration du format de lecture
Par défaut, les colonnes JSON et Variant sont lues sous forme de `VariantType`. Vous pouvez modifier ce comportement pour les lire comme des chaînes :
```scala theme={null}
// Lire JSON/Variant comme des chaînes au lieu de VariantType
spark.conf.set("spark.clickhouse.read.jsonAs", "string")
val df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
// la colonne data sera de type StringType et contiendra des chaînes JSON
```
```python theme={null}
# Lire JSON/Variant comme des chaînes au lieu de VariantType
spark.conf.set("spark.clickhouse.read.jsonAs", "string")
df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
# la colonne data sera de type StringType et contiendra des chaînes JSON
```
```java theme={null}
// Lire JSON/Variant comme des chaînes au lieu de VariantType
spark.conf().set("spark.clickhouse.read.jsonAs", "string");
Dataset df = spark.sql("SELECT id, data FROM clickhouse.default.json_table");
// la colonne data sera de type StringType et contiendra des chaînes JSON
```
### Prise en charge des formats d’écriture
La prise en charge de l’écriture avec VariantType varie selon le format :
| Format | Prise en charge | Remarques |
| ------ | --------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| JSON | ✅ Complète | Prend en charge les types `JSON` et `Variant`. Recommandé pour les données VariantType |
| Arrow | ⚠️ Partielle | Prend en charge l’écriture vers le type `JSON` de ClickHouse. Ne prend pas en charge le type `Variant` de ClickHouse. La prise en charge complète est en attente de résolution de [https://github.com/ClickHouse/ClickHouse/issues/92752](https://github.com/ClickHouse/ClickHouse/issues/92752) |
Configurez le format d’écriture :
```scala theme={null}
spark.conf.set("spark.clickhouse.write.format", "json") // Recommended for Variant types
```
Si vous devez écrire dans un type `Variant` de ClickHouse, utilisez le format JSON. Le format Arrow ne prend en charge l’écriture que dans le type `JSON`.
### Bonnes pratiques
1. **Utilisez le type JSON pour les données exclusivement JSON** : si vous stockez uniquement des objets JSON, utilisez le type JSON par défaut (sans la propriété `variant_types`)
2. **Spécifiez explicitement les types** : lorsque vous utilisez `Variant()`, listez explicitement tous les types que vous prévoyez de stocker
3. **Activez les fonctionnalités expérimentales** : assurez-vous que ClickHouse a bien `allow_experimental_json_type = 1` activé
4. **Utilisez le format JSON pour les écritures** : le format JSON est recommandé pour les données de type VariantType afin d'assurer une meilleure compatibilité
5. **Tenez compte des schémas de requêtes** : les types JSON/Variant prennent en charge les requêtes JSON path de ClickHouse pour un filtrage efficace
6. **Indications de colonne pour les performances** : lorsque vous utilisez des champs JSON dans ClickHouse, l'ajout d'indications de colonne améliore les performances des requêtes. Actuellement, l'ajout d'indications de colonne via Spark n'est pas pris en charge. Consultez l'[issue GitHub #497](https://github.com/ClickHouse/spark-clickhouse-connector/issues/497) pour suivre cette fonctionnalité.
### Exemple : workflow complet
```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);
```
## Configurations
Voici les configurations paramétrables disponibles dans le connecteur.
**Utilisation des configurations** : il s’agit d’options de configuration au niveau de Spark qui s’appliquent à la fois à API Catalog et à TableProvider API. Elles peuvent être définies de deux manières :
1. **Configuration Spark globale** (s’applique à toutes les opérations) :
```python theme={null}
spark.conf.set("spark.clickhouse.write.batchSize", "20000")
spark.conf.set("spark.clickhouse.write.compression.codec", "lz4")
```
2. **Surcharge par opération** (TableProvider API uniquement - peut remplacer les paramètres globaux) :
```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()
```
Vous pouvez également les définir dans `spark-defaults.conf` ou lors de la création de la session Spark.
| Clé | Valeur par défaut | Description | Disponible depuis |
| ------------------------------------------------------------------------ | ---------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------- |
| spark.clickhouse.ignoreUnsupportedTransform | true | ClickHouse prend en charge l’utilisation d’expressions complexes comme clés de sharding ou valeurs de partition, par ex. `cityHash64(col_1, col_2)`, mais Spark ne les prend pas en charge pour le moment. Si `true`, les expressions non prises en charge sont ignorées et un avertissement est consigné ; sinon, l’opération échoue immédiatement en levant une exception. **Avertissement** : lorsque `spark.clickhouse.write.distributed.convertLocal=true`, ignorer des clés de sharding non prises en charge peut corrompre les données. Le connecteur le vérifie et lève une erreur par défaut. Pour l’autoriser, définissez explicitement `spark.clickhouse.write.distributed.convertLocal.allowUnsupportedSharding=true`. | 0.4.0 |
| spark.clickhouse.read.compression.codec | lz4 | Le codec utilisé pour décompresser les données à la lecture. Codecs pris en charge : none, lz4. | 0.5.0 |
| spark.clickhouse.read.distributed.convertLocal | true | Lors de la lecture d’une table Distributed, lisez la table locale au lieu de la table Distributed elle-même. Si `true`, `spark.clickhouse.read.distributed.useClusterNodes` est ignoré. | 0.1.0 |
| spark.clickhouse.read.fixedStringAs | binary | Lit le type FixedString de ClickHouse sous la forme du type de données Spark spécifié. Types pris en charge : binary, string | 0.8.0 |
| spark.clickhouse.read.format | json | Format de sérialisation utilisé pour la lecture. Formats pris en charge : json, binary | 0.6.0 |
| spark.clickhouse.read.runtimeFilter.enabled | false | Activer le filtre runtime pour la lecture. | 0.8.0 |
| spark.clickhouse.read.splitByPartitionId | true | Si `true`, construit le filtre de partition d’entrée à partir de la colonne virtuelle `_partition_id`, au lieu de la valeur de partition. Il existe des problèmes connus lors de la construction des prédicats SQL à partir de la valeur de partition. Cette fonctionnalité nécessite ClickHouse Server v21.6+ | 0.4.0 |
| spark.clickhouse.useNullableQuerySchema | false | Si `true`, marque tous les champs du schéma de la requête comme Nullable lors de l'exécution de `CREATE/REPLACE TABLE ... AS SELECT ...` lors de la création de la table. Remarque : cette configuration nécessite SPARK-43390 (disponible dans Spark 3.5) ; sans ce patch, elle se comporte toujours comme `true`. | 0.8.0 |
| spark.clickhouse.write.batchSize | 10000 | Nombre d’enregistrements par lot lors de l’écriture dans ClickHouse. | 0.1.0 |
| spark.clickhouse.write.compression.codec | lz4 | Le codec utilisé pour compresser les données à l’écriture. Codecs pris en charge : none, lz4. | 0.3.0 |
| spark.clickhouse.write.distributed.convertLocal | false | Lors de l’écriture dans une table Distributed, écrivez dans la table locale plutôt que dans la table Distributed elle-même. Si `true`, ignore `spark.clickhouse.write.distributed.useClusterNodes`. Cela contourne le routage natif de ClickHouse et oblige Spark à évaluer la clé de sharding. Si vous utilisez des expressions de sharding non prises en charge, définissez `spark.clickhouse.ignoreUnsupportedTransform` sur `false` afin d’éviter des erreurs silencieuses dans la distribution des données. | 0.1.0 |
| spark.clickhouse.write.distributed.convertLocal.allowUnsupportedSharding | false | Autorise l’écriture dans des tables Distributed avec `convertLocal=true` et `ignoreUnsupportedTransform=true` lorsque la clé de sharding n’est pas prise en charge. Cette option est dangereuse et peut entraîner une corruption des données en raison d’un sharding incorrect. Lorsqu’elle est définie sur `true`, vous devez vous assurer que vos données sont correctement triées/réparties entre les shards avant l’écriture, car Spark ne peut pas évaluer l’expression de sharding non prise en charge. Ne définissez cette option sur `true` que si vous comprenez les risques et avez vérifié la répartition de vos données. Par défaut, cette combinaison générera une erreur afin d’éviter toute corruption silencieuse des données. | 0.10.0 |
| spark.clickhouse.write.distributed.useClusterNodes | true | Écrire sur tous les nœuds du cluster lors de l’écriture dans une table Distributed. | 0.1.0 |
| spark.clickhouse.write.format | arrow | Format de sérialisation pour l'écriture. Formats pris en charge : json, arrow | 0.4.0 |
| spark.clickhouse.write.localSortByKey | true | Si `true`, effectue un tri local sur les clés de tri avant l’écriture. | 0.3.0 |
| spark.clickhouse.write.localSortByPartition | la valeur de spark.clickhouse.write.repartitionByPartition | Si `true`, effectue un tri local par partition avant l’écriture. Si cette option n’est pas définie, elle est égale à `spark.clickhouse.write.repartitionByPartition`. | 0.3.0 |
| spark.clickhouse.write.maxRetry | 3 | Nombre maximal de tentatives de réécriture pour une écriture par lot unique ayant échoué avec des codes autorisant une nouvelle tentative. | 0.1.0 |
| spark.clickhouse.write.repartitionByPartition | true | Indique s’il faut repartitionner les données selon les clés de partition de la table ClickHouse afin de respecter la distribution de la table ClickHouse avant l’écriture. | 0.3.0 |
| spark.clickhouse.write.repartitionNum | 0 | Les données doivent être repartitionnées avant l’écriture afin de correspondre à la distribution de la table ClickHouse. Utilisez cette configuration pour spécifier le nombre de partitions ; une valeur inférieure à 1 signifie qu’aucun repartitionnement n’est requis. | 0.1.0 |
| spark.clickhouse.write.repartitionStrictly | false | Si `true`, Spark répartit strictement les enregistrements entrants entre les partitions afin de respecter la distribution requise avant de les transmettre à la table de la source de données lors de l’écriture. Sinon, Spark peut appliquer certaines optimisations pour accélérer la requête, mais ne plus respecter cette exigence de distribution. Cette configuration nécessite SPARK-37523 (disponible dans Spark 3.4) ; sans ce correctif, elle se comporte toujours comme si elle était définie sur `true`. | 0.3.0 |
| spark.clickhouse.write.retryInterval | 10s | L’intervalle, en secondes, entre deux nouvelles tentatives d’écriture. | 0.1.0 |
| spark.clickhouse.write.retryableErrorCodes | 241 | Les codes d’erreur pouvant faire l’objet d’une nouvelle tentative, renvoyés par le serveur ClickHouse lorsqu’une écriture échoue. | 0.1.0 |
## Types de données pris en charge
Cette section présente la correspondance des types de données entre Spark et ClickHouse. Les tableaux ci-dessous offrent un aperçu rapide
des conversions de types de données lors de la lecture de données de ClickHouse dans Spark et de l’insertion de données de Spark dans ClickHouse.
### Lecture des données de ClickHouse dans Spark
| Type de données ClickHouse | Type de données Spark | Pris en charge | Est primitif | Remarques |
| ----------------------------------------------------------------- | ------------------------------ | -------------- | ------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `Nothing` | `NullType` | ✅ | Oui | |
| `Bool` | `BooleanType` | ✅ | Oui | |
| `UInt8`, `Int16` | `ShortType` | ✅ | Oui | |
| `Int8` | `ByteType` | ✅ | Oui | |
| `UInt16`,`Int32` | `IntegerType` | ✅ | Oui | |
| `UInt32`,`Int64`, `UInt64` | `LongType` | ✅ | Oui | |
| `Int128`,`UInt128`, `Int256`, `UInt256` | `DecimalType(38, 0)` | ✅ | Oui | |
| `Float32` | `FloatType` | ✅ | Oui | |
| `Float64` | `DoubleType` | ✅ | Oui | |
| `String`, `UUID`, `Enum8`, `Enum16`, `IPv4`, `IPv6` | `StringType` | ✅ | Oui | |
| `FixedString` | `BinaryType`, `StringType` | ✅ | Oui | Contrôlé par la configuration `READ_FIXED_STRING_AS` |
| `Decimal` | `DecimalType` | ✅ | Oui | Précision et échelle jusqu'à `Decimal128` |
| `Decimal32` | `DecimalType(9, scale)` | ✅ | Oui | |
| `Decimal64` | `DecimalType(18, scale)` | ✅ | Oui | |
| `Decimal128` | `DecimalType(38, scale)` | ✅ | Oui | |
| `Date`, `Date32` | `DateType` | ✅ | Oui | |
| `DateTime`, `DateTime32`, `DateTime64` | `TimestampType` | ✅ | Oui | |
| `Array` | `ArrayType` | ✅ | Non | Le type d'élément du tableau est également converti |
| `Map` | `MapType` | ✅ | Non | Les clés sont limitées à `StringType` |
| `IntervalYear` | `YearMonthIntervalType(Year)` | ✅ | Oui | |
| `IntervalMonth` | `YearMonthIntervalType(Month)` | ✅ | Oui | |
| `IntervalDay`, `IntervalHour`, `IntervalMinute`, `IntervalSecond` | `DayTimeIntervalType` | ✅ | Non | Le type d'intervalle spécifique est utilisé |
| `JSON`, `Variant` | `VariantType` | ✅ | Non | Nécessite Spark 4.0+ et ClickHouse 25.3+. Peut être lu comme `StringType` avec `spark.clickhouse.read.jsonAs=string` |
| `Object` | | ❌ | | |
| `Nested` | | ❌ | | |
| `Tuple` | `StructType` | ✅ | Non | Prend en charge les tuples nommés et non nommés. Les tuples nommés sont mappés aux champs de la structure par nom, les tuples non nommés utilisent `_1`, `_2`, etc. Prend en charge les structures imbriquées et les champs nullable |
| `Point` | | ❌ | | |
| `Polygon` | | ❌ | | |
| `MultiPolygon` | | ❌ | | |
| `Ring` | | ❌ | | |
| `IntervalQuarter` | | ❌ | | |
| `IntervalWeek` | | ❌ | | |
| `Decimal256` | | ❌ | | |
| `AggregateFunction` | | ❌ | | |
| `SimpleAggregateFunction` | | ❌ | | |
### Insertion de données de Spark dans ClickHouse
| Type de données Spark | Type de données ClickHouse | Pris en charge | Est primitif | Notes |
| ------------------------------------- | -------------------------- | -------------- | ------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `BooleanType` | `Bool` | ✅ | Oui | Correspond au type `Bool` (et non à `UInt8`) depuis la version 0.9.0 |
| `ByteType` | `Int8` | ✅ | Oui | |
| `ShortType` | `Int16` | ✅ | Oui | |
| `IntegerType` | `Int32` | ✅ | Oui | |
| `LongType` | `Int64` | ✅ | Oui | |
| `FloatType` | `Float32` | ✅ | Oui | |
| `DoubleType` | `Float64` | ✅ | Oui | |
| `StringType` | `String` | ✅ | Oui | |
| `VarcharType` | `String` | ✅ | Oui | |
| `CharType` | `String` | ✅ | Oui | |
| `DecimalType` | `Decimal(p, s)` | ✅ | Oui | Précision et échelle jusqu’à `Decimal128` |
| `DateType` | `Date` | ✅ | Oui | |
| `TimestampType` | `DateTime` | ✅ | Oui | |
| `ArrayType` (liste, tuple ou tableau) | `Array` | ✅ | Non | Le type des éléments du tableau est également converti |
| `MapType` | `Map` | ✅ | Non | Les clés sont limitées à `StringType` |
| `StructType` | `Tuple` | ✅ | Non | Converti en `Tuple` nommé avec des noms de champs. |
| `VariantType` | `JSON` ou `Variant` | ✅ | Non | Nécessite Spark 4.0+ et ClickHouse 25.3+. Le type `JSON` est utilisé par défaut. Utilisez la propriété `clickhouse.column..variant_types` pour spécifier `Variant` avec plusieurs types. |
| `Object` | | ❌ | | |
| `Nested` | | ❌ | | |
## Contribution et support
Si vous souhaitez contribuer au projet ou signaler un problème, vos retours sont les bienvenus !
Consultez notre [repository GitHub](https://github.com/ClickHouse/spark-clickhouse-connector) pour ouvrir une issue, proposer des
améliorations ou soumettre une pull request.
Les contributions sont les bienvenues ! Veuillez consulter les consignes de contribution dans le repository avant de commencer.
Merci de nous aider à améliorer notre ClickHouse Spark connector !