Skip to content
ClickHouse Docs
ClickHouse DocsClickHouse Docs
49.6kGet Started

Conector de Spark

View as MarkdownEdit this pageReport an issueSet up docs MCP
Compatible con ClickHouse

Este conector aprovecha optimizaciones específicas de ClickHouse, como la partición avanzada y el pushdown de predicados, para mejorar el rendimiento de las consultas y el manejo de los datos. El conector se basa en el conector JDBC oficial de ClickHouse y gestiona su propio catálogo.

Antes de Spark 3.0, Spark no tenía un concepto de catálogo integrado, por lo que los usuarios solían recurrir a sistemas de catálogo externos, como Hive Metastore o AWS Glue. Con estas soluciones externas, los usuarios tenían que registrar manualmente las tablas de sus fuentes de datos antes de poder acceder a ellas desde Spark. Sin embargo, desde la introducción del concepto de catálogo en Spark 3.0, Spark ahora puede detectar tablas automáticamente al registrar plugins de catálogo.

El catálogo predeterminado de Spark es spark_catalog, y las tablas se identifican como {catalog name}.{database}.{table}. Con la nueva función de catálogo, ahora es posible añadir y trabajar con varios catálogos en una sola aplicación de Spark.

Elegir entre API de catálogo y TableProvider API

El ClickHouse Spark connector admite dos formas de acceso: la API de catálogo y la TableProvider API (acceso basado en formatos). Comprender las diferencias te ayudará a elegir el enfoque adecuado para tu caso de uso.

API de catálogo vs TableProvider API

Característica API de catálogo TableProvider API
Configuración Centralizada mediante la configuración de Spark Por operación mediante opciones
Detección de tablas Automática mediante el catálogo Especificación manual de la tabla
Operaciones DDL Compatibilidad completa (CREATE, DROP, ALTER) Limitadas (solo creación automática de tablas)
Integración con Spark SQL Nativa (clickhouse.database.table) Requiere especificar el formato
Caso de uso Conexiones estables a largo plazo con configuración centralizada Acceso ad hoc, dinámico o temporal

Requisitos

  • Java 8 o 17 (se requiere Java 17+ para Spark 4.0)
  • Scala 2.12 o 2.13 (Spark 4.0 solo es compatible con Scala 2.13)
  • Apache Spark 3.3, 3.4, 3.5 o 4.0

Matriz de compatibilidad

Versión Versiones compatibles de Spark Versión de ClickHouse JDBC
main Spark 3.3, 3.4, 3.5, 4.0 0.9.4
0.10.0 Spark 3.3, 3.4, 3.5, 4.0 0.9.5
0.9.0 Spark 3.3, 3.4, 3.5, 4.0 0.9.4
0.8.1 Spark 3.3, 3.4, 3.5 0.6.3
0.7.3 Spark 3.3, 3.4 0.4.6
0.6.0 Spark 3.3 0.3.2-patch11
0.5.0 Spark 3.2, 3.3 0.3.2-patch11
0.4.0 Spark 3.2, 3.3 No depende de
0.3.0 Spark 3.2, 3.3 No depende de
0.2.1 Spark 3.2 No depende de
0.1.2 Spark 3.2 No depende de

Instalación y configuración

Para integrar ClickHouse con Spark, hay varias opciones de instalación que se adaptan a distintas configuraciones de proyecto. Puede añadir el ClickHouse Spark connector como dependencia directamente en el archivo de compilación de su proyecto (por ejemplo, en pom.xml para Maven o build.sbt para SBT). Como alternativa, puede colocar los archivos JAR necesarios en la carpeta $SPARK_HOME/jars/ o pasarlos directamente como una opción de Spark mediante la marca --jars en el comando spark-submit. Ambos enfoques garantizan que el conector de ClickHouse esté disponible en su entorno de Spark.

Importar como dependencia

<dependency>
  <groupId>com.clickhouse.spark</groupId>
  <artifactId>clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}</artifactId>
  <version>{{ stable_version }}</version>
</dependency>
<dependency>
  <groupId>com.clickhouse</groupId>
  <artifactId>clickhouse-jdbc</artifactId>
  <classifier>all</classifier>
  <version>{{ clickhouse_jdbc_version }}</version>
  <exclusions>
    <exclusion>
      <groupId>*</groupId>
      <artifactId>*</artifactId>
    </exclusion>
  </exclusions>
</dependency>

Para usar una versión SNAPSHOT, siga las instrucciones de Sonatype para consumir versiones SNAPSHOT con Maven.

Descargar la biblioteca

El patrón de nombres del archivo JAR binario es:

clickhouse-spark-runtime-${spark_binary_version}_${scala_binary_version}-${version}.jar

Puedes encontrar todos los archivos JAR publicados disponibles en el repositorio central de Maven. Los archivos JAR SNAPSHOT de compilación diaria están disponibles a través del repositorio de snapshots de Sonatype configurado anteriormente.

Registrar el catálogo (obligatorio)

Para acceder a sus tablas de ClickHouse, debe configurar un nuevo catálogo de Spark con la siguiente configuración:

Propiedad Valor Valor predeterminado Obligatorio
spark.sql.catalog.<catalog_name> com.clickhouse.spark.ClickHouseCatalog N/A Sí
spark.sql.catalog.<catalog_name>.host <clickhouse_host> localhost No
spark.sql.catalog.<catalog_name>.protocol http http No
spark.sql.catalog.<catalog_name>.http_port <clickhouse_port> 8123 No
spark.sql.catalog.<catalog_name>.user <clickhouse_username> default No
spark.sql.catalog.<catalog_name>.password <clickhouse_password> (cadena vacía) No
spark.sql.catalog.<catalog_name>.database <database> default No
spark.<catalog_name>.write.format json arrow No

Esta configuración se puede establecer de una de las siguientes maneras:

  • Edite o cree spark-defaults.conf.
  • Pase la configuración a su comando spark-submit (o a sus comandos spark-shell/spark-sql de la CLI).
  • Agregue la configuración al iniciar su contexto.

Uso de la TableProvider API (acceso basado en formatos)

Además del enfoque basado en catálogos, el ClickHouse Spark connector admite un patrón de acceso basado en formatos a través de la TableProvider API.

Ejemplo de lectura basada en formato

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# Leer desde ClickHouse con la API de formato
df = spark.read \
    .format("clickhouse") \
    .option("host", "your-clickhouse-host") \
    .option("protocol", "https") \
    .option("http_port", "8443") \
    .option("database", "default") \
    .option("table", "your_table") \
    .option("user", "default") \
    .option("password", "your_password") \
    .option("ssl", "true") \
    .load()

df.show()

Ejemplo de escritura basada en el formato

# Escribe en ClickHouse mediante la API de formato
df.write \
    .format("clickhouse") \
    .option("host", "your-clickhouse-host") \
    .option("protocol", "https") \
    .option("http_port", "8443") \
    .option("database", "default") \
    .option("table", "your_table") \
    .option("user", "default") \
    .option("password", "your_password") \
    .option("ssl", "true") \
    .mode("append") \
    .save()

Funciones de TableProvider

La API de TableProvider ofrece varias funciones avanzadas:

Creación automática de tablas

Al escribir en una tabla inexistente, el conector la crea automáticamente con un esquema adecuado. El conector aplica valores predeterminados inteligentes:

  • Engine: Usa MergeTree() de forma predeterminada si no se especifica. Puede indicar un motor distinto mediante la opción motor (p. ej., ReplacingMergeTree(), SummingMergeTree(), etc.)
  • ORDER BY: Obligatorio: debe especificar explícitamente la opción order_by al crear una tabla nueva. El conector valida que todas las columnas indicadas existan en el esquema.
  • Compatibilidad con claves Nullable: Agrega automáticamente settings.allow_nullable_key=1 si ORDER BY contiene columnas Nullable
# La tabla se creará automáticamente con un ORDER BY explícito (obligatorio)
df.write \
    .format("clickhouse") \
    .option("host", "your-host") \
    .option("database", "default") \
    .option("table", "new_table") \
    .option("order_by", "id") \
    .mode("append") \
    .save()

# Especificar opciones de creación de la tabla con un motor personalizado
df.write \
    .format("clickhouse") \
    .option("host", "your-host") \
    .option("database", "default") \
    .option("table", "new_table") \
    .option("order_by", "id, timestamp") \
    .option("motor", "ReplacingMergeTree()") \
    .option("settings.allow_nullable_key", "1") \
    .mode("append") \
    .save()

Opciones de conexión de TableProvider

Al usar la API basada en formatos, las siguientes opciones de conexión están disponibles:

Opciones de conexión

Opción Descripción Valor predeterminado Requerido
host Hostname del servidor ClickHouse localhost Sí
protocol Protocolo de conexión (http o https) http No
http_port Puerto HTTP/HTTPS 8123 No
database Nombre de la base de datos default Sí
table Nombre de la tabla N/A Sí
user Nombre de usuario para la autenticación default No
password Contraseña para la autenticación (cadena vacía) No
ssl Habilita la conexión SSL false No
ssl_mode Modo SSL (NONE, STRICT, etc.) STRICT No
timezone Zona horaria para operaciones de fecha y hora server No

Opciones de creación de tablas

Estas opciones se usan cuando la tabla no existe y es necesario crearla:

Opción Descripción Valor predeterminado Obligatorio
order_by Columnas que se usarán para la cláusula ORDER BY. Separadas por comas si hay varias columnas N/A Sí
engine Motor de tabla de ClickHouse (p. ej., MergeTree(), ReplacingMergeTree(), SummingMergeTree(), etc.) MergeTree() No
settings.allow_nullable_key Habilita claves anulables en ORDER BY (para ClickHouse Cloud) Detectado automáticamente** No
settings.<key> Cualquier configuración de tabla de ClickHouse N/A No
cluster Nombre del clúster para tablas Distributed N/A No
clickhouse.column.<name>.variant_types Lista separada por comas de tipos de ClickHouse para columnas Variant (p. ej., String, Int64, Bool, JSON). Los nombres de tipo distinguen entre mayúsculas y minúsculas. Los espacios después de las comas son opcionales. N/A No
  • La opción order_by es obligatoria al crear una tabla nueva. Todas las columnas especificadas deben existir en el esquema. ** Se establece automáticamente en 1 si ORDER BY contiene columnas anulables y no se proporciona explícitamente.

Modos de escritura

El conector de Spark (tanto TableProvider API como API de catálogo) admite los siguientes modos de escritura de Spark:

  • append: Añade datos a una tabla existente
  • overwrite: Reemplaza todos los datos de la tabla (trunca la tabla)
# Modo overwrite (primero trunca la tabla)
df.write \
    .format("clickhouse") \
    .option("host", "your-host") \
    .option("database", "default") \
    .option("table", "my_table") \
    .mode("overwrite") \
    .save()

Configuración de las opciones de ClickHouse

Tanto la API de catálogo como la TableProvider API permiten configurar opciones específicas de ClickHouse (no opciones del conector). Estas se transfieren a ClickHouse al crear tablas o ejecutar consultas.

Las opciones de ClickHouse permiten configurar ajustes específicos de ClickHouse, como allow_nullable_key, index_granularity y otros ajustes a nivel de tabla o de consulta. Estas son distintas de las opciones del conector (como host, database, table), que controlan cómo el conector se conecta a ClickHouse.

Uso de la TableProvider API

Con la TableProvider API, use el formato de opción settings.<key>:

df.write \
    .format("clickhouse") \
    .option("host", "your-host") \
    .option("database", "default") \
    .option("table", "my_table") \
    .option("order_by", "id") \
    .option("settings.allow_nullable_key", "1") \
    .option("settings.index_granularity", "8192") \
    .mode("append") \
    .save()

Con la API de catálogo, utilice el formato spark.sql.catalog.<catalog_name>.option.<key> en la configuración de Spark:

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:

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, asegúrese de habilitar SSL y configurar el modo SSL adecuado. Por ejemplo:

spark.sql.catalog.clickhouse.option.ssl        true
spark.sql.catalog.clickhouse.option.ssl_mode   NONE

Leer datos

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<Row> df = spark.sql("select * from clickhouse.default.example_table");

        df.show();

        spark.stop();
    }

Escribir datos

 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<Row> data = Arrays.asList(
                RowFactory.create(1, "Alice"),
                RowFactory.create(2, "Bob")
        );

        // Crear un DataFrame
        Dataset<Row> df = spark.createDataFrame(data, schema);

        df.writeTo("clickhouse.default.example_table").append();

        spark.stop();
    }

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:

USE clickhouse; 

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

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.

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:

// 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()

Escribir datos de VariantType

Puede escribir datos de VariantType en ClickHouse usando tipos de columna JSON o Variant:

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'
  )
""")

Creación de tablas VariantType con Spark SQL

Puede crear tablas VariantType mediante DDL de Spark SQL:

-- Crear tabla con tipo JSON (predeterminado)
CREATE TABLE clickhouse.default.json_table (
  id INT,
  data VARIANT
) USING clickhouse
TBLPROPERTIES (
  'engine' = 'MergeTree()',
  'order_by' = 'id'
)
-- 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:

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:

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:

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:

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:

// 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

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

Configure el formato de escritura:

spark.conf.set("spark.clickhouse.write.format", "json")  // Recomendado para tipos Variant

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 para seguir el estado de esta funcionalidad.

Ejemplo: Flujo de trabajo completo

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)

Configuraciones

Estas son las configuraciones ajustables disponibles en el conector.


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.<name>.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 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!

Navigation