` en la configuración de Spark:
```text theme={null}
spark.sql.catalog.clickhouse.option.allow_nullable_key 1
spark.sql.catalog.clickhouse.option.index_granularity 8192
```
O configúralos al crear las tablas con 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'
)
```
## Configuración de ClickHouse Cloud
Al conectarse a [ClickHouse Cloud](https://clickhouse.com), asegúrese de habilitar SSL y configurar el modo SSL adecuado. Por ejemplo:
```text theme={null}
spark.sql.catalog.clickhouse.option.ssl true
spark.sql.catalog.clickhouse.option.ssl_mode NONE
```
## Leer datos
```java theme={null}
public static void main(String[] args) {
// Crear una sesión de 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;
```
## Escribir datos
**La sobrescritura de particiones no es compatible**: La Catalog API no admite actualmente operaciones de sobrescritura a nivel de partición (por ejemplo, el modo `overwrite` con `partitionBy`). Esta función está en desarrollo. Consulta el [issue n.º 34 en GitHub](https://github.com/ClickHouse/spark-clickhouse-connector/issues/34) para hacer seguimiento de esta función.
```java theme={null}
public static void main(String[] args) throws AnalysisException {
// Crear una sesión de 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 el esquema del 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")
);
// Crear un DataFrame
Dataset df = spark.createDataFrame(data, schema);
df.writeTo("clickhouse.default.example_table").append();
spark.stop();
}
```
```java theme={null}
object NativeSparkWrite extends App {
// Crear una sesión de 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 el esquema del DataFrame
val rows = Seq(Row(1, "John"), Row(2, "Doe"))
val schema = List(
StructField("id", DataTypes.IntegerType, nullable = false),
StructField("name", StringType, nullable = true)
)
// Crear el 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
# Puedes usar cualquier otra combinación de paquetes que cumpla con la matriz de compatibilidad indicada anteriormente.
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")
# Crear DataFrame
data = [Row(id=11, name="John"), Row(id=12, name="Doe")]
df = spark.createDataFrame(data)
# Escribir el DataFrame en ClickHouse
df.writeTo("clickhouse.default.example_table").append()
```
```sql theme={null}
-- resultTable es el DataFrame intermedio de Spark que queremos insertar en clickhouse.default.example_table
INSERT INTO TABLE clickhouse.default.example_table
SELECT * FROM resultTable;
```
## Operaciones DDL
Puede realizar operaciones DDL en su instancia de ClickHouse con Spark SQL, y todos los cambios se conservan de inmediato en
ClickHouse.
Spark SQL le permite escribir consultas exactamente como lo haría en ClickHouse,
por lo que puede ejecutar directamente comandos como CREATE TABLE, TRUNCATE y más, sin modificarlos, por ejemplo:
Al usar Spark SQL, solo se puede ejecutar una instrucción a la 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
);
```
Los ejemplos anteriores muestran consultas de Spark SQL que puedes ejecutar dentro de tu aplicación con cualquier API: Java, Scala,
PySpark o shell.
## Trabajar con VariantType
La compatibilidad con VariantType está disponible en Spark 4.0+ y requiere ClickHouse 25.3+ con los tipos experimentales JSON/Variant habilitados.
El conector admite `VariantType` de Spark para trabajar con datos semiestructurados. VariantType se asigna a los tipos `JSON` y `Variant` de ClickHouse, lo que permite almacenar y consultar de forma eficiente datos con esquemas flexibles.
Esta sección se centra específicamente en el mapeo y uso de VariantType. Para obtener una descripción general completa de todos los tipos de datos compatibles, consulte la sección [Tipos de datos compatibles](#supported-data-types).
### Mapeo de tipos de ClickHouse
| Tipo de ClickHouse | Tipo de Spark | Descripción |
| ---------------------- | ------------- | ---------------------------------------------------------------- |
| `JSON` | `VariantType` | Almacena únicamente objetos JSON (deben empezar con `{`) |
| `Variant(T1, T2, ...)` | `VariantType` | Almacena varios tipos, incluidos tipos primitivos, arrays y JSON |
### Lectura de datos de VariantType
Al leer datos de ClickHouse, las columnas `JSON` y `Variant` se asignan automáticamente a `VariantType` de Spark:
```scala theme={null}
// Leer una columna JSON como VariantType
val df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
// Acceder a los datos de Variant
df.show()
// Convertir Variant a una cadena JSON para su inspección
import org.apache.spark.sql.functions._
df.select(
col("id"),
to_json(col("data")).as("data_json")
).show()
```
```python theme={null}
# Leer una columna JSON como VariantType
df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
# Acceder a los datos de Variant
df.show()
# Convertir Variant a una cadena JSON para su inspección
from pyspark.sql.functions import to_json
df.select(
"id",
to_json("data").alias("data_json")
).show()
```
```java theme={null}
// Leer una columna JSON como VariantType
Dataset df = spark.sql("SELECT id, data FROM clickhouse.default.json_table");
// Acceder a los datos de Variant
df.show();
// Convertir Variant a una cadena JSON para su inspección
import static org.apache.spark.sql.functions.*;
df.select(
col("id"),
to_json(col("data")).as("data_json")
).show();
```
### Escribir datos de VariantType
Puede escribir datos de VariantType en ClickHouse usando tipos de columna JSON o Variant:
```scala theme={null}
import org.apache.spark.sql.functions._
// Crear un DataFrame con datos JSON
val jsonData = Seq(
(1, """{"name": "Alice", "age": 30}"""),
(2, """{"name": "Bob", "age": 25}"""),
(3, """{"name": "Charlie", "city": "NYC"}""")
).toDF("id", "json_string")
// Convertir cadenas JSON a VariantType
val variantDF = jsonData.select(
col("id"),
parse_json(col("json_string")).as("data")
)
// Escribir en ClickHouse con el tipo JSON (solo objetos JSON)
variantDF.writeTo("clickhouse.default.user_data").create()
// O especificar Variant con varios 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
# Crear un DataFrame con datos 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 cadenas JSON a VariantType
variant_df = df.select(
"id",
parse_json("json_string").alias("data")
)
# Escribir en ClickHouse con el tipo JSON
variant_df.writeTo("clickhouse.default.user_data").create()
# O especificar Variant con varios 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.*;
// Crear un DataFrame con datos 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 cadenas JSON a VariantType
Dataset variantDF = jsonDF.select(
col("id"),
parse_json(col("json_string")).as("data")
);
// Escribir en ClickHouse con el tipo JSON (solo objetos JSON)
variantDF.writeTo("clickhouse.default.user_data").create();
// O especificar Variant con varios 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'" +
")");
```
### Creación de tablas VariantType con Spark SQL
Puede crear tablas VariantType mediante DDL de Spark SQL:
```sql theme={null}
-- Crear tabla con tipo JSON (predeterminado)
CREATE TABLE clickhouse.default.json_table (
id INT,
data VARIANT
) USING clickhouse
TBLPROPERTIES (
'engine' = 'MergeTree()',
'order_by' = 'id'
)
```
```sql theme={null}
-- Crear tabla con tipo Variant que admite múltiples 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'
)
```
### Configuración de los tipos Variant
Al crear tablas con columnas VariantType, puede especificar qué tipos de ClickHouse usar:
#### Tipo JSON (predeterminado)
Si no se especifica la propiedad `variant_types`, la columna utiliza de forma predeterminada el tipo `JSON` de ClickHouse, que solo acepta objetos JSON:
```sql theme={null}
CREATE TABLE clickhouse.default.json_table (
id INT,
data VARIANT
) USING clickhouse
TBLPROPERTIES (
'engine' = 'MergeTree()',
'order_by' = 'id'
)
```
Esto crea la siguiente consulta de ClickHouse:
```sql theme={null}
CREATE TABLE json_table (id Int32, data JSON) ENGINE = MergeTree() ORDER BY id
```
#### Tipo Variant con varios tipos
Para admitir tipos primitivos, arrays y objetos JSON, especifique los tipos en la propiedad `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'
)
```
Esto crea la siguiente consulta de 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 compatibles con el `tipo Variant`
Los siguientes tipos de ClickHouse pueden usarse en `Variant()`:
* **Primitivos**: `String`, `Int8`, `Int16`, `Int32`, `Int64`, `UInt8`, `UInt16`, `UInt32`, `UInt64`, `Float32`, `Float64`, `Bool`
* **Arrays**: `Array(T)`, donde T es cualquier tipo compatible, incluidos los arrays anidados
* **JSON**: `JSON` para almacenar objetos JSON
### Configuración del formato de lectura
De forma predeterminada, las columnas JSON y Variant se leen como `VariantType`. Puedes anular este comportamiento para leerlas como cadenas:
```scala theme={null}
// Leer JSON/Variant como cadenas en lugar de VariantType
spark.conf.set("spark.clickhouse.read.jsonAs", "string")
val df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
// la columna data será StringType y contendrá cadenas JSON
```
```python theme={null}
# Leer JSON/Variant como cadenas en lugar de VariantType
spark.conf.set("spark.clickhouse.read.jsonAs", "string")
df = spark.sql("SELECT id, data FROM clickhouse.default.json_table")
# la columna data será StringType y contendrá cadenas JSON
```
```java theme={null}
// Leer JSON/Variant como cadenas en lugar de VariantType
spark.conf().set("spark.clickhouse.read.jsonAs", "string");
Dataset df = spark.sql("SELECT id, data FROM clickhouse.default.json_table");
// la columna data será StringType y contendrá cadenas JSON
```
### Compatibilidad con formatos de escritura
La compatibilidad de escritura de VariantType varía según el formato:
| Formato | Compatibilidad | Notas |
| ------- | -------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| JSON | ✅ Completa | Admite los tipos `JSON` y `Variant`. Recomendado para datos de VariantType |
| Arrow | ⚠️ Parcial | Permite escribir en el tipo `JSON` de ClickHouse. No admite el tipo `Variant` de ClickHouse. La compatibilidad completa está pendiente de que se resuelva [https://github.com/ClickHouse/ClickHouse/issues/92752](https://github.com/ClickHouse/ClickHouse/issues/92752) |
Configure el formato de escritura:
```scala theme={null}
spark.conf.set("spark.clickhouse.write.format", "json") // Recomendado para tipos Variant
```
Si necesita escribir en un tipo `Variant` de ClickHouse, use el formato JSON. El formato Arrow solo permite escribir en el tipo `JSON`.
### Prácticas recomendadas
1. **Use JSON type for JSON-only data**: Si solo almacena objetos JSON, use el tipo JSON predeterminado (sin la propiedad `variant_types`)
2. **Specify types explicitly**: Al usar `Variant()`, enumere explícitamente todos los tipos que piensa almacenar
3. **Enable experimental features**: Asegúrese de que ClickHouse tenga habilitado `allow_experimental_json_type = 1`
4. **Use JSON format for writes**: Se recomienda usar el formato JSON para escribir datos de VariantType, ya que ofrece una mejor compatibilidad
5. **Consider query patterns**: Los tipos JSON/Variant admiten consultas de rutas JSON de ClickHouse para un filtrado eficiente
6. **Column hints for performance**: Al usar campos JSON en ClickHouse, agregar sugerencias de columna mejora el rendimiento de las consultas. Actualmente, no es compatible agregar sugerencias de columna mediante Spark. Consulte el [GitHub issue #497](https://github.com/ClickHouse/spark-clickhouse-connector/issues/497) para seguir el estado de esta funcionalidad.
### Ejemplo: Flujo de trabajo completo
```scala theme={null}
import org.apache.spark.sql.functions._
// Habilitar el tipo JSON experimental en ClickHouse
spark.sql("SET allow_experimental_json_type = 1")
// Crear tabla con columna 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 datos con tipos mixtos
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")
// Convertir a VariantType y escribir
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()
// Leer y 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 el tipo JSON experimental en ClickHouse
spark.sql("SET allow_experimental_json_type = 1")
# Crear tabla con columna 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 datos con tipos mixtos
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"])
# Convertir a VariantType y escribir
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()
# Leer y 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 el tipo JSON experimental en ClickHouse
spark.sql("SET allow_experimental_json_type = 1");
// Crear tabla con columna 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 datos con tipos mixtos
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);
// Convertir a VariantType y escribir
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();
// Leer y 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);
```
## Configuraciones
Estas son las configuraciones ajustables disponibles en el conector.
**Uso de las configuraciones**: Estas son opciones de configuración a nivel de Spark que se aplican tanto a Catalog API como a TableProvider API. Se pueden configurar de dos maneras:
1. **Configuración global de Spark** (se aplica a todas las operaciones):
```python theme={null}
spark.conf.set("spark.clickhouse.write.batchSize", "20000")
spark.conf.set("spark.clickhouse.write.compression.codec", "lz4")
```
2. **Sobrescritura por operación** (solo en TableProvider API; puede sobrescribir la configuración global):
```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, puede configurarlas en `spark-defaults.conf` o al crear la sesión de Spark.
| Clave | Predeterminado | Descripción | Desde |
| ------------------------------------------------------------------------ | ------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------ |
| spark.clickhouse.ignoreUnsupportedTransform | true | ClickHouse admite el uso de expresiones complejas como claves de sharding o valores de partición, p. ej., `cityHash64(col_1, col_2)`, y actualmente Spark no puede admitirlas. Si es `true`, ignora las expresiones no compatibles y registra una advertencia; de lo contrario, falla de inmediato con una excepción. **Advertencia**: Cuando `spark.clickhouse.write.distributed.convertLocal=true`, ignorar claves de sharding no compatibles puede corromper los datos. El conector valida esto y, de forma predeterminada, genera un error. Para permitirlo, establece explícitamente `spark.clickhouse.write.distributed.convertLocal.allowUnsupportedSharding=true`. | 0.4.0 |
| spark.clickhouse.read.compression.codec | lz4 | El codec utilizado para descomprimir los datos durante la lectura. Codecs compatibles: none, lz4. | 0.5.0 |
| spark.clickhouse.read.distributed.convertLocal | true | Al leer una tabla Distributed, lea la tabla local en lugar de la propia tabla. Si es `true`, ignore `spark.clickhouse.read.distributed.useClusterNodes`. | 0.1.0 |
| spark.clickhouse.read.fixedStringAs | binary | Lee el tipo FixedString de ClickHouse como el tipo de datos de Spark especificado. Tipos admitidos: binary, string | 0.8.0 |
| spark.clickhouse.read.format | json | Formato de serialización para la lectura. Formatos admitidos: json, binary | 0.6.0 |
| spark.clickhouse.read.runtimeFilter.enabled | false | Activa el filtro en tiempo de ejecución para la lectura. | 0.8.0 |
| spark.clickhouse.read.splitByPartitionId | true | Si es `true`, construye el filtro de partición de entrada usando la columna virtual `_partition_id`, en lugar del valor de la partición. Se conocen problemas al generar predicados SQL a partir del valor de la partición. Esta función requiere ClickHouse Server v21.6+ | 0.4.0 |
| spark.clickhouse.useNullableQuerySchema | false | Si es `true`, marca todos los campos del esquema de la consulta como anulables al ejecutar `CREATE/REPLACE TABLE ... AS SELECT ...` durante la creación de la tabla. Ten en cuenta que esta configuración requiere SPARK-43390 (disponible en Spark 3.5); sin este parche, siempre actúa como `true`. | 0.8.0 |
| spark.clickhouse.write.batchSize | 10000 | El número de registros por lote al escribir en ClickHouse. | 0.1.0 |
| spark.clickhouse.write.compression.codec | lz4 | El códec utilizado para comprimir los datos durante la escritura. Códecs compatibles: none, lz4. | 0.3.0 |
| spark.clickhouse.write.distributed.convertLocal | false | Al escribir en una tabla Distributed, escriba en la tabla local en lugar de en la propia tabla. Si es `true`, ignore `spark.clickhouse.write.distributed.useClusterNodes`. Esto omite el enrutamiento nativo de ClickHouse, por lo que Spark debe evaluar la clave de sharding. Cuando use expresiones de sharding no compatibles, establezca `spark.clickhouse.ignoreUnsupportedTransform` en `false` para evitar errores silenciosos en la distribución de datos. | 0.1.0 |
| spark.clickhouse.write.distributed.convertLocal.allowUnsupportedSharding | false | Permite escribir en tablas Distributed con `convertLocal=true` e `ignoreUnsupportedTransform=true` cuando la clave de sharding no es compatible. Esto es peligroso y puede provocar corrupción de datos debido a un sharding incorrecto. Cuando se establece en `true`, debe asegurarse de que los datos estén correctamente ordenados/distribuidos por sharding antes de escribirlos, ya que Spark no puede evaluar la expresión de sharding no compatible. Establézcalo en `true` solo si comprende los riesgos y ha verificado la distribución de sus datos. De forma predeterminada, esta combinación producirá un error para evitar la corrupción silenciosa de los datos. | 0.10.0 |
| spark.clickhouse.write.distributed.useClusterNodes | true | Escribir en todos los nodos del clúster al escribir en una tabla Distributed. | 0.1.0 |
| spark.clickhouse.write.format | arrow | Formato de serialización para la escritura. Formatos admitidos: json, arrow | 0.4.0 |
| spark.clickhouse.write.localSortByKey | true | Si es `true`, ordena localmente por las claves de ordenación antes de escribir. | 0.3.0 |
| spark.clickhouse.write.localSortByPartition | valor de spark.clickhouse.write.repartitionByPartition | Si es `true`, ordena localmente por partición antes de escribir. Si no se especifica, equivale a `spark.clickhouse.write.repartitionByPartition`. | 0.3.0 |
| spark.clickhouse.write.maxRetry | 3 | El número máximo de reintentos de escritura que se realizarán para una sola escritura por lotes fallida con códigos que permiten reintento. | 0.1.0 |
| spark.clickhouse.write.repartitionByPartition | true | Si los datos deben reparticionarse por las claves de partición de ClickHouse para ajustarse a la distribución de la tabla de ClickHouse antes de la escritura. | 0.3.0 |
| spark.clickhouse.write.repartitionNum | 0 | Es necesario reparticionar los datos para que coincidan con la distribución de la tabla de ClickHouse antes de escribirlos; use esta configuración para especificar el número de particiones. Un valor inferior a 1 significa que no es necesario hacerlo. | 0.1.0 |
| spark.clickhouse.write.repartitionStrictly | false | Si es `true`, Spark distribuirá estrictamente los registros entrantes entre las particiones para cumplir con la distribución requerida antes de pasarlos a la tabla de la fuente de datos durante la escritura. De lo contrario, Spark puede aplicar ciertas optimizaciones para acelerar la consulta, pero incumplir el requisito de distribución. Ten en cuenta que esta configuración requiere SPARK-37523 (disponible en Spark 3.4); sin este parche, siempre se comporta como `true`. | 0.3.0 |
| spark.clickhouse.write.retryInterval | 10s | El intervalo, en segundos, entre reintentos de escritura. | 0.1.0 |
| spark.clickhouse.write.retryableErrorCodes | 241 | Los códigos de error reintentables devueltos por el servidor ClickHouse cuando falla la escritura. | 0.1.0 |
## Tipos de datos compatibles
Esta sección describe la correspondencia entre los tipos de datos de Spark y ClickHouse. Las tablas siguientes ofrecen una referencia rápida
para convertir tipos de datos al leer desde ClickHouse en Spark y al insertar datos de Spark en ClickHouse.
### Lectura de datos de ClickHouse en Spark
| Tipo de dato de ClickHouse | Tipo de dato de Spark | Compatible | Es primitivo | Notas |
| ----------------------------------------------------------------- | ------------------------------ | ---------- | ------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `Nothing` | `NullType` | ✅ | Sí | |
| `Bool` | `BooleanType` | ✅ | Sí | |
| `UInt8`, `Int16` | `ShortType` | ✅ | Sí | |
| `Int8` | `ByteType` | ✅ | Sí | |
| `UInt16`,`Int32` | `IntegerType` | ✅ | Sí | |
| `UInt32`,`Int64`, `UInt64` | `LongType` | ✅ | Sí | |
| `Int128`,`UInt128`, `Int256`, `UInt256` | `DecimalType(38, 0)` | ✅ | Sí | |
| `Float32` | `FloatType` | ✅ | Sí | |
| `Float64` | `DoubleType` | ✅ | Sí | |
| `String`, `UUID`, `Enum8`, `Enum16`, `IPv4`, `IPv6` | `StringType` | ✅ | Sí | |
| `FixedString` | `BinaryType`, `StringType` | ✅ | Sí | Se controla con la configuración `READ_FIXED_STRING_AS` |
| `Decimal` | `DecimalType` | ✅ | Sí | Precisión y escala de hasta `Decimal128` |
| `Decimal32` | `DecimalType(9, scale)` | ✅ | Sí | |
| `Decimal64` | `DecimalType(18, scale)` | ✅ | Sí | |
| `Decimal128` | `DecimalType(38, scale)` | ✅ | Sí | |
| `Date`, `Date32` | `DateType` | ✅ | Sí | |
| `DateTime`, `DateTime32`, `DateTime64` | `TimestampType` | ✅ | Sí | |
| `Array` | `ArrayType` | ✅ | No | También se convierte el tipo de elemento de Array |
| `Map` | `MapType` | ✅ | No | Las claves se limitan a `StringType` |
| `IntervalYear` | `YearMonthIntervalType(Year)` | ✅ | Sí | |
| `IntervalMonth` | `YearMonthIntervalType(Month)` | ✅ | Sí | |
| `IntervalDay`, `IntervalHour`, `IntervalMinute`, `IntervalSecond` | `DayTimeIntervalType` | ✅ | No | Se usa el tipo de intervalo específico |
| `JSON`, `Variant` | `VariantType` | ✅ | No | Requiere Spark 4.0+ y ClickHouse 25.3+. Puede leerse como `StringType` con `spark.clickhouse.read.jsonAs=string` |
| `Object` | | ❌ | | |
| `Nested` | | ❌ | | |
| `Tuple` | `StructType` | ✅ | No | Admite tanto tuplas con nombre como sin nombre. Las tuplas con nombre se asignan a campos de la estructura por nombre; las tuplas sin nombre usan `_1`, `_2`, etc. Admite structs anidados y campos Nullable |
| `Point` | | ❌ | | |
| `Polygon` | | ❌ | | |
| `MultiPolygon` | | ❌ | | |
| `Ring` | | ❌ | | |
| `IntervalQuarter` | | ❌ | | |
| `IntervalWeek` | | ❌ | | |
| `Decimal256` | | ❌ | | |
| `AggregateFunction` | | ❌ | | |
| `SimpleAggregateFunction` | | ❌ | | |
### Inserción de datos desde Spark en ClickHouse
| Tipo de dato de Spark | Tipo de dato de ClickHouse | Compatible | Es primitivo | Notas |
| ----------------------------------- | -------------------------- | ---------- | ------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `BooleanType` | `Bool` | ✅ | Sí | Se asigna al tipo `Bool` (no `UInt8`) desde la versión 0.9.0 |
| `ByteType` | `Int8` | ✅ | Sí | |
| `ShortType` | `Int16` | ✅ | Sí | |
| `IntegerType` | `Int32` | ✅ | Sí | |
| `LongType` | `Int64` | ✅ | Sí | |
| `FloatType` | `Float32` | ✅ | Sí | |
| `DoubleType` | `Float64` | ✅ | Sí | |
| `StringType` | `String` | ✅ | Sí | |
| `VarcharType` | `String` | ✅ | Sí | |
| `CharType` | `String` | ✅ | Sí | |
| `DecimalType` | `Decimal(p, s)` | ✅ | Sí | Precisión y escala hasta `Decimal128` |
| `DateType` | `Date` | ✅ | Sí | |
| `TimestampType` | `DateTime` | ✅ | Sí | |
| `ArrayType` (list, tuple, or array) | `Array` | ✅ | No | También se convierte el tipo de elemento de Array |
| `MapType` | `Map` | ✅ | No | Las claves están limitadas a `StringType` |
| `StructType` | `Tuple` | ✅ | No | Se convierte en un Tuple con nombre y nombres de campo. |
| `VariantType` | `JSON` o `Variant` | ✅ | No | Requiere Spark 4.0+ y ClickHouse 25.3+. De forma predeterminada, usa el tipo `JSON`. Use la propiedad `clickhouse.column..variant_types` para especificar `Variant` con varios tipos. |
| `Object` | | ❌ | | |
| `Nested` | | ❌ | | |
## Contribuciones y soporte
Si quieres contribuir al proyecto o informar de algún problema, ¡agradecemos tus aportaciones!
Visita nuestro [repositorio de GitHub](https://github.com/ClickHouse/spark-clickhouse-connector) para abrir una incidencia, proponer
mejoras o enviar un pull request.
¡Las contribuciones son bienvenidas! Consulta las directrices de contribución del repositorio antes de empezar.
¡Gracias por ayudarnos a mejorar nuestro ClickHouse Spark connector!