Connecteur Flink

Il s’agit du connecteur sink Apache Flink officiel pris en charge par ClickHouse. Il est construit à l’aide de l’AsyncSinkBase de Flink et du client Java officiel de ClickHouse.

Le connecteur prend en charge la DataStream API d’Apache Flink. La prise en charge de la Table API est prévue pour une prochaine version.

• Java 11+ (pour Flink 1.17+) ou 17+ (pour Flink 2.0+)
• Apache Flink 1.17+

Le connecteur est réparti en deux artefacts afin de prendre en charge à la fois Flink 1.17+ et Flink 2.0+. Choisissez l’artefact correspondant à la version de Flink souhaitée :

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

Le modèle de nommage du JAR binaire est le suivant :

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

où :

• flink_version vaut soit 2.0.0, soit 1.17
• stable_version est une version stable de l’artefact

Vous trouverez tous les fichiers JAR publiés dans le Maven Central Repository.

Supposons que vous vouliez insérer des données CSV brutes dans ClickHouse :

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

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

    // Créer le sink et définir le format à l'aide de `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);

    // Enfin, connectez votre flux DataStream au 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);
}

Vous trouverez d'autres exemples et extraits de code dans nos tests :

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

Nous avons créé un exemple basé sur Maven pour démarrer facilement avec le sink ClickHouse :

• Flink 1.17+
• Flink 2.0.0+

Pour des instructions plus détaillées, consultez le guide des exemples

options et serverSettings doivent être transmis au client sous forme de Map<String, String>. Une map vide pour l'un ou l'autre utilisera respectivement les valeurs par défaut du client ou du serveur.

Par exemple :

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
);

Les options suivantes proviennent directement de AsyncSinkBase de Flink :

Le tableau ci-dessous donne un aperçu rapide de la conversion des types de données lors de l’insertion de données de Flink dans ClickHouse.

Remarques :

• Un ZoneId doit être fourni lors de l'exécution d'opérations sur des dates.
• La précision et l'échelle doivent être fournies lors de l'exécution d'opérations décimales.
• Pour que ClickHouse puisse interpréter une String Java comme du JSON, vous devez activer enableJsonSupportAsString dans ClickHouseClientConfig.
• Le connecteur nécessite un ElementConvertor pour faire correspondre les éléments du DataStream d'entrée aux payloads ClickHouse. À cette fin, le connecteur fournit ClickHouseConvertor et POJOConvertor, que vous pouvez utiliser pour implémenter ce mapping à l'aide des méthodes de sérialisation de DataWriter ci-dessus.

Vous trouverez la liste des formats d’entrée ClickHouse disponibles sur cette page de documentation et dans ClickHouseFormat.java.

Pour spécifier le format que le connecteur doit utiliser pour sérialiser votre DataStream en payloads ClickHouse, utilisez la fonction setClickHouseFormat. Par exemple :

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);

Le connecteur expose les métriques supplémentaires suivantes, en plus des métriques existantes de Flink :

• Le sink fournit actuellement une garantie de livraison at-least-once. Les travaux visant à prendre en charge une sémantique exactly-once sont suivis ici.
• Le sink ne prend pas encore en charge de file dead-letter (DLQ) pour la mise en mémoire tampon des enregistrements qui ne peuvent pas être traités. En attendant, le connector tentera de réinsérer les enregistrements en échec et les abandonnera en cas d'échec. Cette fonctionnalité est suivie ici.
• Le sink ne prend pas encore en charge la création via la Table API de Flink ou Flink SQL. Cette fonctionnalité est suivie ici.

• Le connecteur est testé quotidiennement, via un workflow CI, avec un ensemble de versions récentes de ClickHouse, y compris latest et head. Les versions testées sont mises à jour périodiquement, à mesure que de nouvelles versions de ClickHouse sont publiées. Consultez ici la liste des versions testées chaque jour pour le connecteur.
• Consultez la politique de sécurité de ClickHouse pour connaître les vulnérabilités de sécurité connues et savoir comment signaler une vulnérabilité.
• Nous vous recommandons de mettre régulièrement à niveau le connecteur afin de ne pas manquer les correctifs de sécurité et les nouvelles améliorations.
• Si vous rencontrez un problème de migration, veuillez créer une issue GitHub et nous vous répondrons !

• Pour des performances optimales, assurez-vous que le type d’élément de votre DataStream n’est pas un type générique ; voir ici pour la distinction des types dans Flink. Les éléments non génériques évitent le surcoût de sérialisation induit par Kryo et améliorent le débit vers ClickHouse.
• Nous recommandons de définir maxBatchSize à au moins 1000, et idéalement entre 10 000 et 100 000. Consultez ce guide sur les insertions en masse pour plus d’informations.
• Pour effectuer une déduplication de type OLTP ou un upsert dans ClickHouse, reportez-vous à cette page de documentation. Remarque : à ne pas confondre avec la déduplication par lot qui intervient lors des nouvelles tentatives.

L’erreur suivante peut survenir :

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)

Cause : Le plus souvent, l’erreur CANNOT_READ_ALL_DATA signifie que le schéma de votre table ClickHouse n’est plus aligné sur celui de vos enregistrements Flink. Cela peut se produire si l’un ou l’autre est modifié de manière non rétrocompatible.

Solution : Mettez à jour le schéma de votre table ClickHouse ou le type de données d’entrée du connecteur (ou les deux) afin qu’ils soient compatibles. Si nécessaire, consultez la correspondance de types pour savoir comment faire correspondre les types Java aux types ClickHouse. Remarque : s’il reste encore des enregistrements en transit, vous devrez réinitialiser l’état de Flink lorsque vous redémarrerez le connecteur.

Il se peut que le débit du connecteur n'augmente pas avec le parallélisme du job (nombre de tâches Flink) lors de l'écriture dans ClickHouse.

Cause : le processus de fusion des parts en arrière-plan de ClickHouse peut ralentir les inserts. Cela peut se produire lorsque la taille de lot configurée est trop faible, que le connecteur effectue des flushs trop fréquents, ou en raison d'une combinaison des deux.

Solution : surveillez les métriques numRequestSubmitted et actualRecordsPerBatch afin de déterminer comment ajuster la taille de lot (maxBatchSize) et la fréquence des flushs. Consultez également Utilisation avancée et recommandée pour des recommandations sur le dimensionnement des lots.

Cause : un ou plusieurs lots ont été abandonnés, soit à cause d’un échec non récupérable, soit parce qu’ils n’ont pas pu être insérés dans le nombre de tentatives configuré (définissable via ClickHouseClientConfig.setNumberOfRetries()). Remarque : par défaut, le connecteur tentera de réinsérer un lot jusqu’à 3 fois avant de l’abandonner.

Solution : inspectez les logs du TaskManager et/ou les pile(s) d’appels pour identifier la cause du problème.

Si vous souhaitez contribuer au projet ou signaler un problème, nous serons ravis de recevoir votre contribution ! Rendez-vous sur notre dépôt GitHub pour ouvrir une issue, suggérer des améliorations ou soumettre une pull request.

Les contributions sont les bienvenues ! Veuillez consulter le guide de contribution dans le dépôt avant de commencer. Merci de nous aider à améliorer le connecteur Flink pour ClickHouse !