Leggere le informazioni sullo stato di Structured Streaming

È possibile usare le operazioni dataframe o le funzioni con valori di tabella SQL per eseguire query sui dati e sui metadati dello stato structured streaming. Usare queste funzioni per osservare le informazioni sullo stato per le query con stato Structured Streaming, che possono essere utili per il monitoraggio e il debug.

È necessario avere accesso in lettura al percorso del checkpoint per una query in streaming al fine di eseguire query sui dati di stato o sui metadati. Le funzioni descritte in questo articolo forniscono l'accesso in sola lettura ai dati e ai metadati di stato. È possibile usare solo la semantica di lettura batch per eseguire query sulle informazioni sullo stato.

Requisiti

• Usare una delle configurazioni di calcolo seguenti:
  • Databricks Runtime 16.3 e versioni successive nel calcolo configurato con la modalità di accesso standard.
  • Databricks Runtime 14.3 LTS e versioni successive nel calcolo configurato con modalità di accesso dedicato o senza isolamento.
• Accesso in lettura al percorso del checkpoint usato dalla query di streaming.

Leggere l'archivio di stato di Structured Streaming

È possibile leggere le informazioni sull'archivio di stato per le query Structured streaming eseguite in qualsiasi runtime di Databricks supportato. Usare la sintassi seguente:

Pitone

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

Scala

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

SQL

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

Opzioni dell'API lettore di stato e schema

Per un elenco completo delle opzioni di formato statestore, vedi Archivio di stato.

I dati di output hanno lo schema seguente:

In Databricks Runtime 16.4 LTS e versioni successive, quando l'opzione readChangeFeed è impostata su true, i dati di output hanno lo schema seguente:

Vedere read_statestore funzione con valori di tabella.

Leggere le modifiche dello stato di Structured Streaming

Disponibile in Databricks Runtime 16.4 LTS e versioni successive. Per leggere il modo in cui lo stato cambia tra microbatches invece di visualizzare lo stato completo in un singolo microbatch, impostare readChangeFeed su true e specificare changeStartBatchId. Facoltativamente, specificare changeEndBatchId. Per un elenco completo delle opzioni, consulta State store.

Ad esempio, per leggere le modifiche dello stato dal batch 2 tramite il batch di cui è stato eseguito il commit più recente:

Pitone

df = (spark.read
  .format("statestore")
  .option("readChangeFeed", True)
  .option("changeStartBatchId", 2)
  .load("<checkpointLocation>")
)

Scala

val df = spark.read
  .format("statestore")
  .option("readChangeFeed", true)
  .option("changeStartBatchId", 2)
  .load("<checkpointLocation>")

SQL

SELECT * FROM read_statestore(
    '<checkpointLocation>',
    readChangeFeed => true,
    changeStartBatchId => 2
);

Lo schema di output include colonne aggiuntive batch_id e change_type. Per lo schema completo, vedere Opzioni e schema dell'API di lettura dello stato.

Leggere i metadati di stato di Streaming strutturato

Disponibile in Databricks Runtime 14.3 LTS o versione successiva. È possibile leggere le informazioni sui metadati di stato per le query structured streaming:

Pitone

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

Scala

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

SQL

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

I dati restituiti hanno lo schema seguente:

Vedere read_state_metadata funzione con valori di tabella.

Esempio: Eseguire una query su un lato di un join tra flussi

Usare la sintassi seguente per eseguire una query sul lato sinistro di un join tra flussi.

Pitone

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

Scala

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

SQL

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

Esempio: Eseguire query nell'archivio di stato per il flusso con più operatori con stato

Questo esempio usa il lettore di metadati di stato per raccogliere i dettagli dei metadati di una query di streaming con più operatori con stato, quindi usa i risultati dei metadati come opzioni per il lettore di stato.

Il lettore di metadati di stato accetta il percorso del checkpoint come unica opzione, come nell'esempio di sintassi seguente:

Pitone

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

Scala

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

SQL

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

La tabella seguente rappresenta un esempio di output dei metadati dell'archivio di stato.

Per ottenere i risultati per l'operatore dedupeWithinWatermark , eseguire una query sul lettore di stato con l'opzione operatorId , come nell'esempio seguente:

Pitone

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

Scala

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

SQL

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