DataWorks:Source de données Kafka

Module de données d'enregistrement Kafka

Type de données traité

key

Dépend de l'élément de configuration keyType dans la tâche de synchronisation des données. Pour plus d'informations sur le paramètre keyType, consultez la description complète des paramètres en annexe.

value

Dépend de l'élément de configuration valueType dans la tâche de synchronisation des données. Pour plus d'informations sur le paramètre valueType, consultez la description complète des paramètres en annexe.

offset

Long

timestamp

Long

headers

String

partition

Long

Type de tâche de synchronisation

Format de la valeur écrite dans Kafka

Type de champ source

Méthode de traitement pour les opérations d'écriture

Synchronisation hors ligne

Nœud de synchronisation hors ligne dans DataStudio

json

String

Chaîne encodée en UTF-8

Boolean

Converti en chaîne encodée en UTF-8 « true » ou « false »

Heure/Date

Chaîne encodée en UTF-8 au format yyyy-MM-dd HH:mm:ss

Numérique

Chaîne numérique encodée en UTF-8

Flux d'octets

Le flux d'octets est traité comme une chaîne encodée en UTF-8 et converti en chaîne.

text

String

Chaîne encodée en UTF-8

Boolean

Converti en chaîne encodée en UTF-8 « true » ou « false »

Heure/Date

Chaîne encodée en UTF-8 au format yyyy-MM-dd HH:mm:ss

Numérique

Chaîne numérique encodée en UTF-8

Flux d'octets

Le flux d'octets est traité comme une chaîne encodée en UTF-8 et converti en chaîne.

Synchronisation en temps réel : ETL en temps réel vers Kafka

Nœud de synchronisation en temps réel dans DataStudio

json

String

Chaîne encodée en UTF-8

Boolean

Type Boolean JSON

Heure/Date

• Pour les valeurs temporelles avec une précision inférieure à la milliseconde : Converties en un entier JSON de 13 chiffres représentant l'horodatage en millisecondes.

• Pour les valeurs temporelles avec une précision en microsecondes ou nanosecondes : Converties en un nombre à virgule flottante JSON comprenant un entier de 13 chiffres pour l'horodatage en millisecondes et une décimale de 6 chiffres pour l'horodatage en nanosecondes.

Numérique

Type numérique JSON

Flux d'octets

Le flux d'octets est encodé en Base64, puis converti en une chaîne encodée en UTF-8.

text

String

Chaîne encodée en UTF-8

Boolean

Converti en chaîne encodée en UTF-8 « true » ou « false »

Heure/Date

Chaîne encodée en UTF-8 au format yyyy-MM-dd HH:mm:ss

Numérique

Chaîne numérique encodée en UTF-8

Flux d'octets

Le flux d'octets est encodé en Base64, puis converti en une chaîne encodée en UTF-8.

Synchronisation en temps réel : Synchronisation en temps réel d'une base de données entière vers Kafka

Synchronisation en temps réel des données incrémentielles uniquement

Format JSON intégré

String

Chaîne encodée en UTF-8

Boolean

Type Boolean JSON

Heure/Date

Horodatage en millisecondes sur 13 chiffres

Numérique

Valeur numérique JSON

Flux d'octets

Le flux d'octets est encodé en Base64, puis converti en une chaîne encodée en UTF-8.

Solution de synchronisation : Synchronisation en temps réel en un clic vers Kafka

Synchronisation hors ligne complète + synchronisation en temps réel incrémentielle

Format JSON intégré

String

Chaîne encodée en UTF-8

Boolean

Type Boolean JSON

Heure/Date

Horodatage en millisecondes sur 13 chiffres

Numérique

Valeur numérique JSON

Flux d'octets

Le flux d'octets est encodé en Base64, puis converti en une chaîne encodée en UTF-8.

Configuration item

Description

Logon module

Doit être défini sur org.apache.kafka.common.security.plain.PlainLoginModul

username

Nom d'utilisateur. Configurez cet élément selon vos besoins.

password

Mot de passe. Configurez cet élément selon vos besoins.

Parameter

Description

Required

datasource

Nom de la source de données. L'éditeur de code permet d'ajouter des sources de données. La valeur de ce paramètre doit correspondre exactement au nom de la source ajoutée.

Yes

server

Adresse du broker Kafka, au format ip:port.

Vous ne pouvez configurer qu'un seul server, mais vous devez vous assurer que DataWorks peut se connecter aux adresses IP de tous les brokers du cluster Kafka.

Yes

topic

Topic Kafka. Un topic représente un regroupement de flux de messages traités par Kafka.

Yes

column

Données Kafka à lire. Les colonnes constantes, les colonnes de données et les colonnes d'attributs sont prises en charge.

• Colonne constante : colonne entourée de guillemets simples, par exemple ["'abc'", "'123'"].

• Colonne de données

  • Si vos données sont au format JSON, vous pouvez extraire les propriétés de l'objet JSON, par exemple ["event_id"].

  • Si vos données sont au format JSON, vous pouvez extraire les sous-propriétés imbriquées de l'objet JSON, par exemple ["tag.desc"].

• Colonne d'attribut

  • __key__ : clé du message.

  • __value__ : contenu intégral du message.

  • __partition__ : partition dans laquelle se trouve le message actuel.

  • __headers__ : en-têtes du message actuel.

  • __offset__ : décalage (offset) du message actuel.

  • __timestamp__ : horodatage du message actuel.

  L'exemple suivant illustre une configuration complète.

  "column": [
      "__key__",
      "__value__",
      "__partition__",
      "__offset__",
      "__timestamp__",
      "'123'",
      "event_id",
      "tag.desc"
      ]

Yes

keyType

Type de la clé Kafka. Valeurs possibles : BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG et SHORT.

No

valueType

Type de la valeur Kafka. Valeurs possibles : BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG et SHORT.

No

beginDateTime

Heure de début de la consommation des données. Ce paramètre définit la borne inférieure (inclusive) de la plage temporelle. Il s'agit d'une chaîne au format yyyymmddhhmmss. Vous pouvez l'utiliser conjointement avec les scheduling parameters. Pour plus d'informations, consultez la rubrique Formats pris en charge pour les paramètres de planification.

Vous devez spécifier ce paramètre ou beginOffset.

endDateTime

Heure de fin de la consommation des données. Ce paramètre définit la borne supérieure (exclusive) de la plage temporelle. Il s'agit d'une chaîne au format yyyymmddhhmmss. Vous pouvez l'utiliser conjointement avec les scheduling parameters. Pour plus d'informations, consultez la rubrique Formats pris en charge pour les paramètres de planification.

Vous devez spécifier ce paramètre ou endOffset.

beginOffset

Décalage (offset) de départ pour la consommation des données. Vous pouvez le configurer selon les modalités suivantes :

• Un nombre, par exemple 15553274, qui indique le décalage de départ pour la consommation.

• seekToBeginning : indique que la consommation commence à partir du décalage le plus ancien.

• seekToLast : lit les données à partir du décalage enregistré pour l'ID de groupe spécifié par le paramètre group.id dans kafkaConfig. Notez que le client valide automatiquement le décalage du groupe sur le serveur Kafka à intervalles réguliers. Par conséquent, si une tâche échoue et est relancée, des duplications ou des pertes de données peuvent survenir. Si le paramètre skipExceedRecord est défini sur true, la tâche peut ignorer les dernières lignes lues. Le décalage du groupe correspondant à ces données ignorées ayant déjà été validé sur le serveur, ces données ne pourront pas être relues lors de la prochaine exécution de la tâche.

• seekToEnd : indique que la consommation commence à partir du décalage le plus récent. Cela entraînera la lecture de données vides.

Vous devez spécifier ce paramètre ou beginDateTime.

endOffset

Décalage (offset) de fin pour la consommation des données. Ce paramètre permet de contrôler l'arrêt de la tâche de consommation.

Vous devez spécifier ce paramètre ou endDateTime.

skipExceedRecord

Kafka utilise la méthode public ConsumerRecords<K, V> poll(final Duration timeout) pour consommer les données. Un seul appel à poll peut récupérer des données au-delà du endOffset ou de l'endDateTime spécifié. Ce paramètre détermine si ces données excédentaires sont écrites dans la destination. Étant donné que la tâche utilise la validation automatique des décalages, nous recommandons les configurations suivantes :

• Pour les versions de Kafka antérieures à 0.10.2 : définissez skipExceedRecord sur false.

• Pour Kafka 0.10.2 et versions ultérieures : définissez skipExceedRecord sur true.

No. La valeur par défaut est false.

partition

Un topic Kafka comporte plusieurs partitions (partition). Par défaut, une tâche de synchronisation de données lit les données sur une plage de décalages couvrant toutes les partitions du topic. Vous pouvez également spécifier une partition pour limiter la lecture à la plage de décalages d'une seule partition.

No. Aucune valeur par défaut.

kafkaConfig

Lors de la création d'un client KafkaConsumer pour la consommation de données, vous pouvez spécifier des paramètres étendus tels que bootstrap.servers, auto.commit.interval.ms et session.timeout.ms. Utilisez kafkaConfig pour contrôler le comportement de consommation du KafkaConsumer.

No

encoding

Lorsque keyType ou valueType est défini sur STRING, l'encodage spécifié par ce paramètre est utilisé pour analyser la chaîne.

No. La valeur par défaut est UTF-8.

waitTIme

Durée maximale, en secondes, pendant laquelle l'objet consumer attend pour extraire des données de Kafka lors d'une tentative unique.

No. La valeur par défaut est 60.

stopWhenPollEmpty

Les valeurs valides sont true et false. Si ce paramètre est défini sur true et que le consumer extrait des données vides de Kafka (généralement parce que toutes les données du topic ont été lues, ou en raison de problèmes réseau ou de disponibilité du cluster Kafka), la tâche s'arrête immédiatement. Sinon, elle réessaie jusqu'à ce que des données soient à nouveau lues.

No. La valeur par défaut est true.

stopWhenReachEndOffset

Ce paramètre prend effet uniquement lorsque stopWhenPollEmpty est défini sur true. Les valeurs valides sont true et false.

• Si ce paramètre est défini sur true et que le consumer ne reçoit aucune donnée lors d'une requête poll, il vérifie si le décalage le plus récent a été atteint dans les partitions du topic. Si le décalage le plus récent a été atteint pour toutes les partitions, la tâche s'arrête immédiatement. Sinon, elle continue d'essayer d'extraire des données du topic.

• Si ce paramètre est défini sur false et que le consumer ne reçoit aucune donnée lors d'une requête poll, il n'effectue pas cette vérification et arrête la tâche immédiatement.

No. La valeur par défaut est false.

Parameter

Description

fetch.min.bytes

Quantité minimale de données, en octets, que le consumer récupère auprès du broker lors d'une seule requête. Le broker attend que cette quantité de données soit disponible avant de répondre au consumer.

fetch.max.wait.ms

Durée maximale, en millisecondes, pendant laquelle le broker attend que des données deviennent disponibles avant de répondre à une requête de récupération. La valeur par défaut est 500. Le broker répond dès que la condition fetch.min.bytes ou fetch.max.wait.ms est satisfaite.

max.partition.fetch.bytes

Spécifie le nombre maximal d'octets que le broker peut renvoyer au consumer pour chaque partition. La valeur par défaut est 1 Mo.

session.timeout.ms

Spécifie la durée pendant laquelle le consumer peut rester déconnecté du serveur avant d'arrêter de recevoir des services. La valeur par défaut est 30 secondes.

auto.offset.reset

Action entreprise par le consumer lors d'une lecture sans décalage ou avec un décalage invalide (par exemple, si le consumer est resté inactif longtemps et que l'enregistrement correspondant au décalage a expiré et a été supprimé). La valeur par défaut est none, ce qui signifie que le décalage n'est pas réinitialisé automatiquement. Vous pouvez la modifier en earliest, ce qui signifie que le consumer lit les enregistrements de la partition à partir du décalage le plus ancien.

max.poll.records

Nombre de messages pouvant être renvoyés par un seul appel à la méthode poll.

key.deserializer

Méthode de désérialisation pour la clé du message, par exemple org.apache.kafka.common.serialization.StringDeserializer.

value.deserializer

Méthode de désérialisation pour la valeur des données, par exemple org.apache.kafka.common.serialization.StringDeserializer.

ssl.truststore.location

Chemin d'accès au certificat racine SSL.

ssl.truststore.password

Mot de passe du magasin de certificats racines. Si vous utilisez Alibaba Cloud Kafka, définissez cette valeur sur KafkaOnsClient.

security.protocol

Protocole d'accès. Actuellement, seul le protocole SASL_SSL est pris en charge.

sasl.mechanism

Méthode d'authentification SASL. Si vous utilisez Alibaba Cloud Kafka, utilisez PLAIN.

java.security.auth.login.config

Chemin d'accès au fichier d'authentification SASL.

Parameter

Description

Required

datasource

Nom de la source de données. L'éditeur de code prend en charge l'ajout de sources de données. La valeur de ce paramètre doit correspondre au nom de la source de données ajoutée.

Oui

server

Adresse du serveur Kafka au format ip:port.

Oui

topic

Rubrique Kafka. Il s'agit d'une catégorie pour les différents flux de messages traités par Kafka.

Chaque message publié sur un cluster Kafka appartient à une catégorie, appelée rubrique (topic). Une rubrique est un ensemble de messages.

Oui

valueIndex

Colonne utilisée comme valeur dans le writer Kafka. Si ce paramètre n'est pas spécifié, toutes les colonnes sont concaténées par défaut pour former la valeur. Le séparateur est défini par le paramètre fieldDelimiter.

Non

writeMode

Lorsque le paramètre valueIndex n'est pas configuré, ce paramètre détermine le format de concaténation de toutes les colonnes de l'enregistrement source pour former la valeur de l'enregistrement Kafka. Les valeurs valides sont text et JSON. La valeur par défaut est text.

• Si la valeur est text, toutes les colonnes sont concaténées à l'aide du séparateur spécifié par fieldDelimiter.

• Si la valeur est JSON, toutes les colonnes sont concaténées en une chaîne JSON basée sur les noms de champs spécifiés par le paramètre column.

Par exemple, si un enregistrement source comporte trois colonnes avec les valeurs a, b et c, et que writeMode est défini sur text et fieldDelimiter sur #, la valeur de l'enregistrement Kafka écrit est la chaîne a#b#c. Si writeMode est défini sur JSON et column sur [{"name":"col1"},{"name":"col2"},{"name":"col3"}], la valeur de l'enregistrement Kafka écrit est la chaîne {"col1":"a","col2":"b","col3":"c"}.

Si le paramètre valueIndex est configuré, ce paramètre est ignoré.

Non

column

Champs de la table de destination vers lesquels les données doivent être écrites, séparés par des virgules. Par exemple : "column": ["id", "name", "age"].

Lorsque le paramètre valueIndex n'est pas configuré et que writeMode est défini sur JSON, ce paramètre définit les noms de champs dans la structure JSON pour les valeurs de colonne de l'enregistrement source. Par exemple, "column": [{"name":id","type":"JSON_NUMBER"}, {"name":"name","type":"JSON_STRING"}, {"name":"age","type":"JSON_NUMBER"}].

• Si le nombre de colonnes dans l'enregistrement source est supérieur au nombre de noms de champs configurés dans column, les données sont tronquées lors de l'écriture. Par exemple :

  Si un enregistrement source comporte trois colonnes avec les valeurs a, b et c, et que column est configuré comme suit : [{"name":"col1","type":"JSON_STRING"},{"name":"col2","type":"JSON_STRING"}], la valeur de l'enregistrement Kafka écrit est la chaîne {"col1":"a","col2":"b"}.

• Si le nombre de colonnes dans l'enregistrement source est inférieur au nombre de champs spécifiés dans column, les champs supplémentaires sont renseignés avec null ou la chaîne spécifiée par nullValueFormat. Par exemple :

  Si un enregistrement source comporte deux colonnes avec les valeurs a et b, et que column est configuré comme suit : [{"name":"col1","type":"JSON_STRING"},{"name":"col2","type":"JSON_STRING"},{"name":"col3","type":"JSON_STRING"}], la valeur de l'enregistrement Kafka écrit est la chaîne {"col1":"a","col2":"b","col3":null}. Si le paramètre valueIndex est configuré ou si writeMode est défini sur text, ce paramètre est ignoré.

• Si le type de champ JSON n'est pas configuré, le type de champ par défaut est JSON_STRING.

• Pour connaître les valeurs valides du type de champ JSON, consultez la section Annexe : Types de champs JSON.

Si le paramètre valueIndex est configuré ou si writeMode est défini sur text, ce paramètre est ignoré.

Requis lorsque le paramètre valueIndex n'est pas configuré et que writeMode est défini sur JSON.

partition

Spécifie le numéro de la partition de la rubrique Kafka vers laquelle les données sont écrites. Il doit s'agir d'un entier supérieur ou égal à 0.

Non

keyIndex

Colonne utilisée comme clé dans le writer Kafka.

La valeur du paramètre keyIndex doit être un entier supérieur ou égal à 0. Dans le cas contraire, la tâche échouera.

Non

keyIndexes

Tableau des numéros d'ordre des colonnes de l'enregistrement source utilisées comme clé pour l'enregistrement Kafka.

Le numéro d'ordre de la colonne commence à 0. Par exemple, [0,1,2] concatène les valeurs de tous les numéros de colonne configurés avec des virgules pour former la clé de l'enregistrement Kafka. Si ce paramètre n'est pas spécifié, la clé de l'enregistrement Kafka est null et les données sont écrites dans les partitions de la rubrique selon un algorithme round-robin. Vous ne pouvez spécifier que ce paramètre ou keyIndex, mais pas les deux.

Non

fieldDelimiter

Lorsque writeMode est défini sur text et que valueIndex n'est pas configuré, toutes les colonnes de l'enregistrement source sont concaténées à l'aide du séparateur de colonne spécifié par ce paramètre pour former la valeur de l'enregistrement Kafka. Vous pouvez configurer un seul caractère ou plusieurs caractères comme séparateur. Vous pouvez configurer des caractères Unicode au format \u0001. Les caractères d'échappement tels que \t et \n sont pris en charge. La valeur par défaut est \t.

Si writeMode n'est pas défini sur text ou si le paramètre valueIndex est configuré, ce paramètre est ignoré.

Non

keyType

Type de la clé Kafka. Valeurs valides : BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG et SHORT.

Oui

valueType

Type de la valeur Kafka. Valeurs valides : BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG et SHORT.

Oui

nullKeyFormat

Si la valeur de la colonne source spécifiée par keyIndex ou keyIndexes est null, elle est remplacée par la chaîne spécifiée par ce paramètre. Si ce paramètre n'est pas configuré, aucun remplacement n'est effectué.

Non

nullValueFormat

Si la valeur d'une colonne source est null, elle est remplacée par la chaîne spécifiée par ce paramètre lors de l'assemblage de la valeur de l'enregistrement Kafka. Si vous ne spécifiez pas ce paramètre, aucun remplacement n'est effectué.

Non

acks

Configuration acks lors de l'initialisation du producteur Kafka. Elle détermine la méthode d'accusé de réception pour les écritures réussies. Par défaut, le paramètre acks est défini sur all. Les valeurs valides pour acks sont :

• 0 : Aucun accusé de réception pour les écritures réussies.

• 1 : Accusé de réception pour l'écriture réussie sur le réplica principal.

• all : Accusé de réception pour l'écriture réussie sur tous les réplicas.

Non