Lire les informations d'état de Structured Streaming

Vous pouvez utiliser des opérations DataFrame ou des fonctions SQL de valeur de table pour interroger les données et les métadonnées d’état de Structured Streaming. Utilisez ces fonctions pour observer les informations d’état des requêtes stateful dans Structured Streaming, ce qui est utile pour la surveillance et le débogage.

Vous devez disposer d’un accès en lecture au chemin de point de contrôle d’une requête de diffusion en continu pour interroger les métadonnées ou les données d’état. Les fonctions décrites dans cet article fournissent un accès en lecture seule aux métadonnées et aux données d’état. Vous pouvez uniquement utiliser la sémantique de lecture par lots pour interroger les informations d’état.

Spécifications

• Utilisez l’une des configurations de calcul suivantes :
  • Databricks Runtime 16.3 et versions ultérieures sur le calcul configuré avec le mode d’accès standard.
  • Databricks Runtime 14.3 LTS et au-delà sur une infrastructure informatique configurée avec un mode d’accès dédié ou sans isolation.
• Accès en lecture au chemin de point de contrôle utilisé par la requête de diffusion en continu.

Lire le magasin d’état Structured Streaming

Vous pouvez lire les informations de magasin d’état pour les requêtes Structured Streaming exécutées dans n’importe quel Databricks Runtime pris en charge. Utilisez la syntaxe suivante :

Python

df = (spark.read
  .format("statestore")
  .load("/checkpoint/path"))

SQL

SELECT * FROM read_statestore('/checkpoint/path')

Paramètres de l’API Lecteur d’état

L’API lecteur d’état prend en charge les configurations facultatives suivantes :

Les données retournées ont le schéma suivant :

Consultez read_statestoreTVF.

Lire les métadonnées d’état Structured Streaming

Vous pouvez lire les informations de métadonnées d’état pour les requêtes Structured Streaming exécutées dans Databricks Runtime 14.2 ou version ultérieure. Utilisez la syntaxe suivante :

Python

df = (spark.read
  .format("state-metadata")
  .load("<checkpointLocation>"))

SQL

SELECT * FROM read_state_metadata('/checkpoint/path')

Les données retournées ont le schéma suivant :

Consultez read_state_metadataTVF.

Exemple : interroger un côté d’une jointure de flux de flux

Utilisez la syntaxe suivante pour interroger le côté gauche d’une jointure de flux de flux :

Python

left_df = (spark.read
  .format("statestore")
  .option("joinSide", "left")
  .load("/checkpoint/path"))

SQL

SELECT * FROM read_statestore(
    '/checkpoint/path',
    joinSide => 'left'
);

Exemple : magasin d’états de requête pour le flux avec plusieurs opérateurs avec état

Cet exemple utilise le lecteur de métadonnées d’état pour collecter les détails des métadonnées d’une requête de diffusion en continu avec plusieurs opérateurs avec état, puis utilise les résultats des métadonnées comme options pour le lecteur d’état.

Le lecteur de métadonnées d’état prend le chemin de point de contrôle comme seule option, comme dans l’exemple de syntaxe suivant :

Python

df = (spark.read
  .format("state-metadata")
  .load("<checkpointLocation>"))

SQL

SELECT * FROM read_state_metadata('/checkpoint/path')

Le tableau suivant représente un exemple de sortie des métadonnées du magasin d’états :

Pour obtenir les résultats de l’opérateur dedupeWithinWatermark , interrogez le lecteur d’état avec l’option operatorId , comme dans l’exemple suivant :

Python

left_df = (spark.read
  .format("statestore")
  .option("operatorId", 1)
  .load("/checkpoint/path"))

SQL

SELECT * FROM read_statestore(
    '/checkpoint/path',
    operatorId => 1
);