Conector de Flink

Este es el conector sink oficial de Apache Flink compatible con ClickHouse. Está desarrollado con AsyncSinkBase de Flink y el cliente Java oficial de ClickHouse.

El conector es compatible con la API de DataStream de Apache Flink. La compatibilidad con la Table API está prevista para una versión futura.

• Java 11+ (para Flink 1.17+) o 17+ (para Flink 2.0+)
• Apache Flink 1.17+

El conector se distribuye en dos artefactos para admitir tanto Flink 1.17+ como Flink 2.0+. Elige el artefacto que corresponda a la versión de Flink que quieras usar:

<dependency>
    <groupId>com.clickhouse.flink</groupId>
    <artifactId>flink-connector-clickhouse-2.0.0</artifactId>
    <version>{{ stable_version }}</version>
    <classifier>all</classifier>
</dependency>

dependencies {
    implementation("com.clickhouse.flink:flink-connector-clickhouse-2.0.0:{{ stable_version }}")
}

libraryDependencies += "com.clickhouse.flink" % "flink-connector-clickhouse-2.0.0" % {{ stable_version }} classifier "all"

<dependency>
    <groupId>com.clickhouse.flink</groupId>
    <artifactId>flink-connector-clickhouse-1.17</artifactId>
    <version>{{ stable_version }}</version>
    <classifier>all</classifier>
</dependency>

dependencies {
    implementation("com.clickhouse.flink:flink-connector-clickhouse-1.17:{{ stable_version }}")
}

libraryDependencies += "com.clickhouse.flink" % "flink-connector-clickhouse-1.17" % {{ stable_version }} classifier "all"

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

flink-connector-clickhouse-${flink_version}-${stable_version}-all.jar

donde:

• flink_version es uno de 2.0.0 o 1.17
• stable_version es una versión estable del artefacto

Puedes encontrar todos los archivos JAR publicados disponibles en el Repositorio Central de Maven.

Supongamos que quieres insertar datos CSV sin procesar en ClickHouse:

public static void main(String[] args) {
    // Configurar ClickHouseClient
    ClickHouseClientConfig clientConfig = new ClickHouseClientConfig(url, username, password, database, tableName);

    // Crear un ElementConverter
    ElementConverter<String, ClickHousePayload> convertorString = new ClickHouseConvertor<>(String.class);

    // Crear el sink y establecer el formato con `setClickHouseFormat`
    ClickHouseAsyncSink<String> csvSink = new ClickHouseAsyncSink<>(
            convertorString,
            MAX_BATCH_SIZE,
            MAX_IN_FLIGHT_REQUESTS,
            MAX_BUFFERED_REQUESTS,
            MAX_BATCH_SIZE_IN_BYTES,
            MAX_TIME_IN_BUFFER_MS,
            MAX_RECORD_SIZE_IN_BYTES,
            clientConfig
    );

    csvSink.setClickHouseFormat(ClickHouseFormat.CSV);

    // Por último, conecta tu DataStream al sink.
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    Path csvFilePath = new Path(fileFullName);
    FileSource<String> csvSource = FileSource
            .forRecordStreamFormat(new TextLineInputFormat(), csvFilePath)
            .build();

    env.fromSource(
            csvSource,
            WatermarkStrategy.noWatermarks(),
            "GzipCsvSource"
    ).sinkTo(csvSink);
}

Puedes encontrar más ejemplos y fragmentos de código en nuestras pruebas:

• flink-connector-clickhouse-1.17
• flink-connector-clickhouse-2.0.0

Hemos creado un ejemplo basado en Maven para facilitar el inicio con el sink de ClickHouse:

• Flink 1.17+
• Flink 2.0.0+

Para obtener instrucciones más detalladas, consulta la Guía del ejemplo

options y serverSettings deben pasarse al cliente como Map<String, String>. Si cualquiera de los dos es un mapa vacío, se usarán los valores predeterminados del cliente o del servidor, respectivamente.

Por ejemplo:

Map<String, String> javaClientOptions = Map.of(
    ClientConfigProperties.CA_CERTIFICATE.getKey(), "<my_CA_cert>",
    ClientConfigProperties.SSL_CERTIFICATE.getKey(), "<my_SSL_cert>",
    ClientConfigProperties.CLIENT_NETWORK_BUFFER_SIZE.getKey(), "30000",
    ClientConfigProperties.HTTP_MAX_OPEN_CONNECTIONS.getKey(), "5"
);

Map<String, String> serverSettings = Map.of(
    "insert_deduplicate", "1"
);

ClickHouseClientConfig clientConfig = new ClickHouseClientConfig(
    url,
    username,
    password,
    database,
    tableName,
    javaClientOptions,
    serverSettings,
    false // enableJsonSupportAsString
);

Las siguientes opciones provienen directamente de AsyncSinkBase de Flink:

La siguiente tabla ofrece una referencia rápida para la conversión de tipos de datos al insertar datos desde Flink en ClickHouse.

Notas:

• Se debe proporcionar un ZoneId al realizar operaciones con fechas.
• Se deben proporcionar la precisión y la escala al realizar operaciones decimales.
• Para que ClickHouse pueda interpretar un String de Java como JSON, es necesario habilitar enableJsonSupportAsString en ClickHouseClientConfig.
• El conector requiere un ElementConvertor para asignar los elementos del DataStream de entrada a los payloads de ClickHouse. Para ello, el conector proporciona ClickHouseConvertor y POJOConvertor, que pueden usarse para implementar esta asignación mediante los métodos de serialización de DataWriter indicados anteriormente.

Puedes consultar la lista de formatos de entrada de ClickHouse disponibles en esta página de documentación y en ClickHouseFormat.java.

Para especificar el formato que debe usar el conector para serializar tu DataStream como payloads de ClickHouse, utiliza la función setClickHouseFormat. Por ejemplo:

ClickHouseAsyncSink<String> csvSink = new ClickHouseAsyncSink<>(
        convertorString,
        MAX_BATCH_SIZE,
        MAX_IN_FLIGHT_REQUESTS,
        MAX_BUFFERED_REQUESTS,
        MAX_BATCH_SIZE_IN_BYTES,
        MAX_TIME_IN_BUFFER_MS,
        MAX_RECORD_SIZE_IN_BYTES,
        clientConfig
);
csvSink.setClickHouseFormat(ClickHouseFormat.CSV);

El conector expone las siguientes métricas adicionales, además de las que ya ofrece Flink:

• El sink actualmente ofrece una garantía de entrega de al menos una vez. El trabajo para lograr la semántica exactly-once se está rastreando aquí.
• El sink todavía no admite una cola de mensajes fallidos (DLQ) para el almacenamiento en búfer de registros que no se pueden procesar. Mientras tanto, el conector intentará volver a insertar los registros que fallen y los descartará si no lo consigue. Esta funcionalidad se está rastreando aquí.
• El sink todavía no admite la creación mediante la Table API de Flink ni con Flink SQL. Esta funcionalidad se está rastreando aquí.

• El conector se prueba a diario, mediante un flujo de trabajo de CI, con varias versiones recientes de ClickHouse, incluidas latest y head. Las versiones probadas se actualizan periódicamente a medida que entran en uso nuevas versiones de ClickHouse. Consulta aquí las versiones con las que se prueba el conector cada día.
• Consulta la política de seguridad de ClickHouse para conocer las vulnerabilidades de seguridad conocidas y cómo informar de una vulnerabilidad.
• Recomendamos actualizar el conector de forma continua para no perder ninguna corrección de seguridad ni mejora nueva.
• Si tienes algún problema con la migración, crea un issue en GitHub y te responderemos.

• Para obtener un rendimiento óptimo, asegúrese de que el tipo de elemento de su DataStream no sea un tipo genérico; consulte aquí la distinción de tipos de Flink. Los elementos no genéricos evitan la sobrecarga de serialización que introduce Kryo y mejoran el rendimiento hacia ClickHouse.
• Recomendamos configurar maxBatchSize con un valor mínimo de 1000 y, preferiblemente, entre 10,000 y 100,000. Consulte esta guía sobre inserciones masivas para obtener más información.
• Para realizar deduplicación de estilo OLTP o upsert en ClickHouse, consulte esta página de documentación. Nota: no debe confundirse con la deduplicación por lotes que se produce en los reintentos.

Puede aparecer el siguiente error:

com.clickhouse.client.api.ServerException: Code: 33. DB::Exception: Cannot read all data. Bytes read: 9205. Bytes expected: 1100022.: (at row 9) : While executing BinaryRowInputFormat. (CANNOT_READ_ALL_DATA)

Causa: Lo más habitual es que el error CANNOT_READ_ALL_DATA signifique que el esquema de tu tabla de ClickHouse ha dejado de coincidir con el esquema de tus registros de Flink. Esto puede ocurrir cuando uno de los dos se modifica de una forma no compatible con versiones anteriores.

Solución: Actualiza el esquema de tu tabla de ClickHouse o el tipo de datos de entrada del conector (o ambos) para que sean compatibles. Si es necesario, consulta la correspondencia de tipos para ver cómo se asignan los tipos de Java a los tipos de ClickHouse. Nota: si todavía hay registros en tránsito, tendrás que restablecer el estado de Flink al reiniciar el conector.

Es posible que el rendimiento del conector no escale con el paralelismo del job (número de tareas de Flink) al escribir en ClickHouse.

Causa: El proceso de fusión de partes en segundo plano de ClickHouse puede estar ralentizando las inserciones. Esto puede ocurrir cuando el tamaño del lote configurado es demasiado pequeño, el conector hace flush con demasiada frecuencia, o por una combinación de ambos factores.

Solución: Supervise las métricas numRequestSubmitted y actualRecordsPerBatch para determinar cómo ajustar el tamaño del lote (maxBatchSize) y con qué frecuencia hacer flush. Consulte también Uso avanzado y recomendado para ver recomendaciones sobre el tamaño de los lotes.

Causa: Los lotes se descartaron, ya sea por un error no reintentable o porque no pudieron insertarse dentro del número de reintentos configurado (se puede configurar mediante ClickHouseClientConfig.setNumberOfRetries()). Nota: de forma predeterminada, el conector intentará volver a insertar un lote hasta 3 veces antes de descartarlo.

Solución: Inspeccione los logs de TaskManager y/o las trazas de pila para identificar la causa raíz.

Si deseas contribuir al proyecto o informar de algún problema, ¡agradecemos tus aportaciones! Visita nuestro repositorio de GitHub para abrir una issue, proponer mejoras o enviar un pull request.

¡Las contribuciones son bienvenidas! Consulta la guía de contribución del repositorio antes de empezar. ¡Gracias por ayudarnos a mejorar el conector de ClickHouse para Flink!