Intégration entre Apache Beam et ClickHouse

Apache Beam est un modèle de programmation unifié open source qui permet aux développeurs de définir et d’exécuter des pipelines de traitement de données à la fois par lots et en flux continu. La flexibilité d’Apache Beam réside dans sa capacité à prendre en charge un large éventail de scénarios de traitement de données, allant des opérations ETL (Extract, Transform, Load) au traitement complexe d’événements et à l’analyse en temps réel. Cette intégration s’appuie sur le connecteur JDBC officiel de ClickHouse pour la couche d’insertion sous-jacente.

Le paquet d’intégration nécessaire pour intégrer Apache Beam et ClickHouse est maintenu et développé dans Apache Beam I/O Connectors — un bundle d’intégrations pour de nombreux systèmes de stockage de données et bases de données populaires. L’implémentation org.apache.beam.sdk.io.clickhouse.ClickHouseIO se trouve dans le dépôt Apache Beam.

Ajoutez la dépendance suivante à votre outil de gestion de paquets :

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-clickhouse</artifactId>
    <version>${beam.version}</version>
</dependency>

Les artifacts sont disponibles dans le dépôt Maven officiel.

L'exemple suivant lit un fichier CSV nommé input.csv sous forme de PCollection, le convertit en objet Row (à l'aide du schéma défini) et l'insère dans une instance ClickHouse locale à l'aide de ClickHouseIO :

package org.example;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.clickhouse.ClickHouseIO;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.Row;
import org.joda.time.DateTime;

public class Main {

    public static void main(String[] args) {
        // Create a Pipeline object.
        Pipeline p = Pipeline.create();

        Schema SCHEMA =
                Schema.builder()
                        .addField(Schema.Field.of("name", Schema.FieldType.STRING).withNullable(true))
                        .addField(Schema.Field.of("age", Schema.FieldType.INT16).withNullable(true))
                        .addField(Schema.Field.of("insertion_time", Schema.FieldType.DATETIME).withNullable(false))
                        .build();

        // Apply transforms to the pipeline.
        PCollection<String> lines = p.apply("ReadLines", TextIO.read().from("src/main/resources/input.csv"));

        PCollection<Row> rows = lines.apply("ConvertToRow", ParDo.of(new DoFn<String, Row>() {
            @ProcessElement
            public void processElement(@Element String line, OutputReceiver<Row> out) {

                String[] values = line.split(",");
                Row row = Row.withSchema(SCHEMA)
                        .addValues(values[0], Short.parseShort(values[1]), DateTime.now())
                        .build();
                out.output(row);
            }
        })).setRowSchema(SCHEMA);

        rows.apply("Write to ClickHouse",
                        ClickHouseIO.write("jdbc:clickhouse://localhost:8123/default?user=default&password=******", "test_table"));

        // Run the pipeline.
        p.run().waitUntilFinish();
    }
}

Vous pouvez ajuster la configuration de ClickHouseIO.Write à l’aide des fonctions setter suivantes :

Veuillez tenir compte des limitations suivantes lors de l'utilisation du connecteur :

• À ce jour, seule l'opération Sink est prise en charge. Le connecteur ne prend pas en charge l'opération Source.
• ClickHouse effectue une déduplication lors de l'insertion dans une table ReplicatedMergeTree ou dans une table Distributed basée sur une table ReplicatedMergeTree. Sans réplication, l'insertion dans une table MergeTree standard peut entraîner des doublons si une insertion échoue puis aboutit lors d'une nouvelle tentative. Cependant, chaque bloc est inséré de manière atomique, et la taille du bloc peut être configurée à l'aide de ClickHouseIO.Write.withMaxInsertBlockSize(long). La déduplication s'appuie sur les sommes de contrôle des blocs insérés. Pour plus d'informations sur la déduplication, veuillez consulter Déduplication et Configuration de la déduplication à l'insertion.
• Le connecteur n'exécute aucune instruction DDL ; par conséquent, la table cible doit exister avant toute insertion.

• Documentation de la classe ClickHouseIO.
• Dépôt GitHub d'exemples clickhouse-beam-connector.