`read_kafka` fonction table

Article
04/18/2024

S’applique à : coche marquée oui Databricks SQL case marquée oui Databricks Runtime 13.3 LTS et versions ultérieures

Lit les données d’un cluster Apache Kafka et retourne les données sous forme tabulaire.

Peut lire des données d’une ou plusieurs rubriques Kafka. Il prend en charge à la fois les requêtes par lots et l’ingestion en diffusion en continu.

Syntaxe

read_kafka([option_key => option_value ] [, ...])

Arguments

Cette fonction nécessite un appel de paramètre nommé.

option_key : le nom de l’option à configurer. Vous devez utiliser des accents graves (`) pour les options qui contiennent des points (.).
option_value : une expression constante pour définir l’option. Accepte les fonctions littérales et scalaires.

Retours

Enregistrements lus à partir d’un cluster Apache Kafka avec le schéma suivant :

key BINARY : la clé de l’enregistrement Kafka.
value BINARY NOT NULL : la valeur de l’enregistrement Kafka.
topic STRING NOT NULL : le nom de la rubrique Kafka à partir de laquelle l’enregistrement est lu.
partition INT NOT NULL : l’ID de la partition Kafka à partir de laquelle l’enregistrement est lu.
offset BIGINT NOT NULL : le numéro de décalage de l’enregistrement dans TopicPartition Kafka.
timestamp TIMESTAMP NOT NULL : la valeur de temps pour l’enregistrement. La colonne timestampType définit à quoi correspond ce timestamp.
timestampType INTEGER NOT NULL : le type de timestamp spécifié dans la colonne timestamp.
headers ARRAY<STRUCT<key: STRING, VALUE: BINARY>> : valeurs d’en-tête fournies dans le cadre de l’enregistrement (si cette option est activée).

Exemples

-- A batch query to read from a topic.
> SELECT value::string as value
  FROM read_kafka(
    bootstrapServers => 'kafka_server:9092',
    subscribe => 'events'
  ) LIMIT 10;

-- A more advanced query with security credentials for Kafka.
> SELECT * FROM read_kafka(
    bootstrapServers => 'kafka_server:9092',
    subscribe => 'events',
    startingOffsets => 'earliest',
    `kafka.security.protocol` => 'SASL_SSL',
    `kafka.sasl.mechanism` => 'PLAIN',
    `kafka.sasl.jaas.config` =>  'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="{USER_NAME}" password="{PASSWORD}";',
  );

-- Streaming ingestion from Kafka with JSON parsing.
> CREATE OR REFRESH STREAMING TABLE catalog.schema.raw_events AS
  SELECT
    value::string:events,                 -- extract the field `events`
    to_timestamp(value::string:ts) as ts  -- extract the field `ts` and cast to timestamp
  FROM STREAM read_kafka(
    bootstrapServers => 'kafka_server:9092',
    subscribe => 'events'
  );

Options

Vous trouverez une liste détaillée des options dans la documentation Apache Spark.

Options requises

Fournissez l’option ci-dessous pour vous connecter à votre cluster Kafka.

Option
`bootstrapServers` Entrez : `String` Liste séparée par des virgules de paires hôte/port pointant vers le cluster Kafka. Valeur par défaut : Aucun

Fournissez une seule des options ci-dessous pour configurer les rubriques Kafka à partir desquelles extraire des données.

Option
`assign` Entrez : `String` Chaîne JSON qui contient les partitions de rubrique spécifiques à utiliser. Par exemple, pour `'{"topicA":[0,1],"topicB":[2,4]}'`, les partitions 0 et 1 de la rubriqueA seront utilisées. Valeur par défaut : aucune
`subscribe` Entrez : `String` Liste séparée par des virgules des rubriques Kafka à lire. Valeur par défaut : aucune
`subscribePattern` Entrez : `String` Expression régulière correspondant aux rubriques auxquelles s’abonner. Valeur par défaut : Aucun

Options diverses

read_kafka peut être utilisé dans les requêtes par lots et dans les requêtes de diffusion en continu. Les options ci-dessous spécifient le type de requête auquel elles s’appliquent.

Option
`endingOffsets` Type : Type de requête `String` : lot uniquement Décalages à lire pour une requête par lots, soit `"latest"` pour spécifier les enregistrements les plus récents, soit une chaîne JSON spécifiant un décalage de fin pour chaque TopicPartition. Dans JSON, `-1` peut être utilisé pour indiquer le décalage le plus récent. `-2` (le plus ancien) car un décalage n’est pas autorisé. Valeur par défaut : `"latest"`
`endingOffsetsByTimestamp` Type : Type de requête `String` : lot uniquement Chaîne JSON spécifiant un timestamp de fin à lire pour chaque TopicPartition. Les timestamps doivent être fournis en tant que valeur longue du timestamp en millisecondes depuis `1970-01-01 00:00:00 UTC`, par exemple `1686444353000`. Consultez la note ci-dessous pour plus d’informations sur le comportement avec les timestamps. `endingOffsetsByTimestamp` est prioritaire par rapport à `endingOffsets`. Valeur par défaut : Aucun
`endingTimestamp` Type : Type de requête `String` : lot uniquement Valeur de chaîne du timestamp en millisecondes depuis `1970-01-01 00:00:00 UTC`, par exemple `"1686444353000"`. Si Kafka ne retourne pas le décalage correspondant, le décalage est défini sur le plus récent. Consultez la note ci-dessous pour plus d’informations sur le comportement avec les timestamps. Note : `endingTimestamp` est prioritaire sur `endingOffsetsByTimestamp` et `endingOffsets`. Valeur par défaut : aucune
`includeHeaders` Type : Type de requête `Boolean` : diffusion en continu et lot Indique s’il faut inclure les en-têtes Kafka dans la ligne. Valeur par défaut : `false`
`kafka.<consumer_option>` Type : Type de requête `String` : diffusion en continu et lot Toutes les options spécifiques au consommateur Kafka peuvent être transmises avec le préfixe `kafka.`. Ces options doivent être entourées d’accents graves lorsqu’elles sont fournies. Sinon, vous obtiendrez une erreur d’analyseur. Vous trouverez les options dans la documentation Kafka. Note : vous ne devez pas définir les options suivantes avec cette fonction : `key.deserializer`, `value.deserializer`, `bootstrap.servers`, `group.id` Valeur par défaut : Aucun
`maxOffsetsPerTrigger` Type : Type de requête `Long` : diffusion en continu uniquement Limite de débit sur le nombre maximal de décalages ou lignes traités par intervalle de déclencheur. Le nombre total de décalages spécifié sera réparti de manière proportionnelle entre TopicPartitions. Valeur par défaut : Aucun
`startingOffsets` Type : Type de requête `String` : diffusion en continu et lot Point de départ lors du démarrage d’une requête, soit `"earliest"` qui correspond aux décalages les plus anciens, `"latest"` qui correspond aux décalages les plus récents, soit une chaîne JSON spécifiant un décalage de départ pour chaque TopicPartition. Dans JSON, `-2` peut être utilisé pour indiquer le décalage le plus ancien et `-1` le plus récent. Note : pour les requêtes par lot, le décalage le plus récent (implicitement ou à l’aide de -1 dans JSON) n’est pas autorisé. Pour les requêtes de diffusion en continu, cela s’applique uniquement lorsqu’une nouvelle requête est démarrée. Les requêtes de diffusion en continu redémarrées continuent à partir des décalages définis dans le point de contrôle de la requête. Les partitions nouvellement découvertes au cours d’une requête démarreront au plus tôt. Valeur par défaut : `"latest"` pour la diffusion en continu, `"earliest"` pour le lot
`startingOffsetsByTimestamp` Type : Type de requête `String` : diffusion en continu et lot Chaîne JSON spécifiant un timestamp de départ pour chaque TopicPartition. Les timestamps doivent être fournis en tant que valeur longue du timestamp en millisecondes depuis `1970-01-01 00:00:00 UTC`, par exemple `1686444353000`. Consultez la note ci-dessous pour plus d’informations sur le comportement avec les timestamps. Si Kafka ne retourne pas le décalage correspondant, le comportement suit la valeur de l’option `startingOffsetsByTimestampStrategy`. `startingOffsetsByTimestamp` est prioritaire par rapport à `startingOffsets`. Note : Pour les requêtes de diffusion en continu, cela s’applique uniquement lorsqu’une nouvelle requête est démarrée. Les requêtes de diffusion en continu redémarrées continuent à partir des décalages définis dans le point de contrôle de la requête. Les partitions nouvellement découvertes au cours d’une requête démarreront au plus tôt. Valeur par défaut : Aucun
`startingOffsetsByTimestampStrategy` Type : Type de requête `String` : diffusion en continu et lot Cette stratégie est utilisée lorsque le décalage de départ spécifié par timestamp (global ou par partition) ne correspond pas au décalage Kafka retourné. Les stratégies disponibles sont les suivantes : * `"error"` : échec de la requête * `"latest"` : attribue le décalage le plus récent pour ces partitions afin que Spark puisse lire les enregistrements plus récents à partir de ces partitions dans des micro-lots ultérieurs. Valeur par défaut : `"error"`
`startingTimestamp` Type : Type de requête `String` : diffusion en continu et lot Valeur de chaîne du timestamp en millisecondes depuis `1970-01-01 00:00:00 UTC`, par exemple `"1686444353000"`. Consultez la note ci-dessous pour plus d’informations sur le comportement avec les timestamps. Si Kafka ne retourne pas le décalage correspondant, le comportement suit la valeur de l’option `startingOffsetsByTimestampStrategy`. `startingTimestamp` est prioritaire sur `startingOffsetsByTimestamp` et `startingOffsets`. Note : Pour les requêtes de diffusion en continu, cela s’applique uniquement lorsqu’une nouvelle requête est démarrée. Les requêtes de diffusion en continu redémarrées continuent à partir des décalages définis dans le point de contrôle de la requête. Les partitions nouvellement découvertes au cours d’une requête démarreront au plus tôt. Valeur par défaut : Aucun

Remarque

Le décalage retourné pour chaque partition est le décalage le plus ancien dont le timestamp est supérieur ou égal au timestamp donné dans la partition correspondante. Le comportement varie selon les options si Kafka ne retourne pas le décalage correspondant. Veuillez vérifier la description de chaque option.

Spark transmet simplement les informations du timestamp à KafkaConsumer.offsetsForTimes, et n’interprète pas ou ne raisonne pas sur la valeur. Pour en savoir plus sur KafkaConsumer.offsetsForTimes, consultez la documentation. En outre, la signification du timestamp ici peut varier en fonction de la configuration Kafka (log.message.timestamp.type). Pour plus d’informations, consultez la documentation Apache Kafka.

read_kafka fonction table