Connecteur Flink

Introduction à Apache Flink avec ClickHouse

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.

​

Prérequis

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

​

Matrice de compatibilité des versions de Flink

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 :

Version de Flink	Artefact	Version du client Java ClickHouse	Java requis
dernière	flink-connector-clickhouse-2.0.0	0.9.5	Java 17+
2.0.1	flink-connector-clickhouse-2.0.0	0.9.5	Java 17+
2.0.0	flink-connector-clickhouse-2.0.0	0.9.5	Java 17+
1.20.2	flink-connector-clickhouse-1.17	0.9.5	Java 11+
1.19.3	flink-connector-clickhouse-1.17	0.9.5	Java 11+
1.18.1	flink-connector-clickhouse-1.17	0.9.5	Java 11+
1.17.2	flink-connector-clickhouse-1.17	0.9.5	Java 11+

Le connecteur n’a pas été testé avec des versions de Flink antérieures à la version 1.17.2.

​

Installation et configuration

​

Ajouter comme dépendance

​

Pour Flink 2.0+

Maven

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

Gradle

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

SBT

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

​

Pour Flink 1.17+

Maven

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

Gradle

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

SBT

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

​

Télécharger le binaire

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.

​

Utiliser la DataStream API

​

Extrait

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

Java

public static void main(String[] args) {
    // Configurer ClickHouseClient
    ClickHouseClientConfig clickHouseClientConfig = 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,
            clickHouseClientConfig
    );

    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

​

Exemple de prise en main rapide

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 de connexion de DataStream API

​

Options du client ClickHouse

Paramètres	Description	Valeur par défaut	Obligatoire
url	URL ClickHouse complète	N/A	Oui
username	Nom d’utilisateur de la base de données ClickHouse	N/A	Oui
password	Mot de passe de la base de données ClickHouse	N/A	Oui
database	Nom de la base de données ClickHouse	N/A	Oui
table	Nom de la table ClickHouse	N/A	Oui
options	Map des options de configuration du client Java	Map vide	Non
serverSettings	Map des paramètres de session du serveur ClickHouse	Map vide	Non
enableJsonSupportAsString	Paramètre du serveur ClickHouse indiquant qu’une String au format JSON est attendue pour le type de données JSON	true	Non

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.

Toutes les options disponibles du client Java sont répertoriées dans ClientConfigProperties.java et sur cette page de documentation.

Tous les paramètres de session disponibles côté serveur sont répertoriés sur cette page de documentation.

Par exemple :

Java

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 clickHouseClientConfig = new ClickHouseClientConfig(
    url,
    username,
    password,
    database,
    tableName,
    javaClientOptions,
    serverSettings,
    false // enableJsonSupportAsString
);

​

Options du sink

Les options suivantes proviennent directement de AsyncSinkBase de Flink :

Paramètres	Description	Valeur par défaut	Obligatoire
maxBatchSize	Nombre maximal d’enregistrements insérés dans un seul lot	N/A	Oui
maxInFlightRequests	Nombre maximal de requêtes simultanées autorisées avant que le sink n’applique une backpressure	N/A	Oui
maxBufferedRequests	Nombre maximal d’enregistrements pouvant être mis en tampon dans le sink avant l’application d’une backpressure	N/A	Oui
maxBatchSizeInBytes	Taille maximale (en octets) qu’un lot peut atteindre. Tous les lots envoyés seront inférieurs ou égaux à cette taille	N/A	Oui
maxTimeInBufferMS	Durée maximale pendant laquelle un enregistrement peut rester dans le sink avant d’être flushé	N/A	Oui
maxRecordSizeInBytes	Taille maximale d’enregistrement acceptée par le sink ; les enregistrements plus volumineux seront automatiquement rejetés	N/A	Oui

​

Types de données pris en charge

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.

​

Insertion de données depuis Flink dans ClickHouse

Type Java	Type ClickHouse	Pris en charge	Méthode de sérialisation
byte/Byte	Int8	✅	DataWriter.writeInt8
short/Short	Int16	✅	DataWriter.writeInt16
int/Integer	Int32	✅	DataWriter.writeInt32
long/Long	Int64	✅	DataWriter.writeInt64
BigInteger	Int128	✅	DataWriter.writeInt128
BigInteger	Int256	✅	DataWriter.writeInt256
short/Short	UInt8	✅	DataWriter.writeUInt8
int/Integer	UInt8	✅	DataWriter.writeUInt8
int/Integer	UInt16	✅	DataWriter.writeUInt16
long/Long	UInt32	✅	DataWriter.writeUInt32
long/Long	UInt64	✅	DataWriter.writeUInt64
BigInteger	UInt64	✅	DataWriter.writeUInt64
BigInteger	UInt128	✅	DataWriter.writeUInt128
BigInteger	UInt256	✅	DataWriter.writeUInt256
BigDecimal	Decimal	✅	DataWriter.writeDecimal
BigDecimal	Decimal32	✅	DataWriter.writeDecimal
BigDecimal	Decimal64	✅	DataWriter.writeDecimal
BigDecimal	Decimal128	✅	DataWriter.writeDecimal
BigDecimal	Decimal256	✅	DataWriter.writeDecimal
float/Float	Float	✅	DataWriter.writeFloat32
double/Double	Double	✅	DataWriter.writeFloat64
boolean/Boolean	Boolean	✅	DataWriter.writeBoolean
String	String	✅	DataWriter.writeString
String	FixedString	✅	DataWriter.writeFixedString
LocalDate	Date	✅	DataWriter.writeDate
LocalDate	Date32	✅	DataWriter.writeDate32
LocalDateTime	DateTime	✅	DataWriter.writeDateTime
ZonedDateTime	DateTime	✅	DataWriter.writeDateTime
LocalDateTime	DateTime64	✅	DataWriter.writeDateTime64
ZonedDateTime	DateTime64	✅	DataWriter.writeDateTime64
int/Integer	Time	❌	N/A
long/Long	Time64	❌	N/A
byte/Byte	Enum8	✅	DataWriter.writeInt8
int/Integer	Enum16	✅	DataWriter.writeInt16
java.util.UUID	UUID	✅	DataWriter.writeIntUUID
String	JSON	✅	DataWriter.writeJSON
Array<Type>	Array<Type>	✅	DataWriter.writeArray
Map<K,V>	Map<K,V>	✅	DataWriter.writeMap
Tuple<Type,..>	Tuple<T1,T2,..>	✅	DataWriter.writeTuple
Object	Variant	❌	N/A

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.

​

Formats d’entrée pris en charge

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,
        clickHouseClientConfig
);
csvSink.setClickHouseFormat(ClickHouseFormat.CSV);

Par défaut, le connecteur utilisera RowBinaryWithDefaults ou RowBinary si setSupportDefault dans ClickHouseClientConfig est explicitement défini sur true ou false, respectivement.

​

Métriques

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

Métrique	Description	Type	Statut
numBytesSend	Nombre total d’octets envoyés à ClickHouse dans le payload de la requête. Remarque : cette métrique mesure la taille des données sérialisées envoyées sur le réseau et peut différer de written_bytes dans system.query_log de ClickHouse, qui reflète le nombre réel d’octets écrits dans le stockage après traitement	Counter	✅
numRecordSend	Nombre total d’enregistrements envoyés à ClickHouse	Counter	✅
numRequestSubmitted	Nombre total de requêtes envoyées (nombre réel de flush effectués)	Counter	✅
numOfDroppedBatches	Nombre total de lots abandonnés en raison d’échecs non récupérables	Counter	✅
numOfDroppedRecords	Nombre total d’enregistrements abandonnés en raison d’échecs non récupérables	Counter	✅
totalBatchRetries	Nombre total de nouvelles tentatives de lot en raison d’échecs récupérables	Counter	✅
writeLatencyHistogram	Histogramme de la distribution de la latence des écritures réussies (ms)	Histogram	✅
writeFailureLatencyHistogram	Histogramme de la distribution de la latence des écritures en échec (ms)	Histogram	✅
triggeredByMaxBatchSizeCounter	Nombre total de flush déclenchés par l’atteinte de maxBatchSize	Counter	✅
triggeredByMaxBatchSizeInBytesCounter	Nombre total de flush déclenchés par l’atteinte de maxBatchSizeInBytes	Counter	✅
triggeredByMaxTimeInBufferMSCounter	Nombre total de flush déclenchés par l’atteinte de maxTimeInBufferMS	Counter	✅
actualRecordsPerBatch	Histogramme de la distribution de la taille réelle des lots	Histogram	✅
actualBytesPerBatch	Histogramme de la distribution du nombre réel d’octets par lot	Histogram	✅

​

Limitations

• 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.

​

Compatibilité des versions de ClickHouse et sécurité

• 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 !

​

Utilisation avancée recommandée

• 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.

​

Dépannage

​

CANNOT_READ_ALL_DATA

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.

​

Faible débit

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.

​

Il manque des lignes dans ma table ClickHouse

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.

​

Contribuer et obtenir de l’aide

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 !