Čtení informací o stavu strukturovaného streamování

Operace datového rámce nebo funkce tabulek SQL můžete použít k dotazování dat a metadat stavu strukturovaného streamování. Pomocí těchto funkcí můžete sledovat stavové informace pro stavové dotazy strukturovaného streamování, které můžou být užitečné pro monitorování a ladění.

Abyste mohli dotazovat stavová data nebo metadata, musíte mít přístup ke čtení cesty ke kontrolnímu bodu dotazu streamování. Funkce popsané v tomto článku poskytují přístup jen pro čtení k datům a metadatům stavu. K dotazování informací o stavu můžete použít pouze sémantiku dávkového čtení.

Poznámka:

Nelze dotazovat informace o stavu pro deklarativní kanály Lakeflow Sparku, streamované tabulky nebo materializovaná zobrazení. Informace o stavu nemůžete dotazovat pomocí výpočetních prostředků bez serveru nebo výpočetních prostředků nakonfigurovaných pomocí standardního režimu přístupu.

Požadavky

• Použijte jednu z následujících konfigurací výpočetních prostředků:
  • Databricks Runtime 16.3 a novější na výpočetních prostředcích nakonfigurovaných pomocí standardního režimu přístupu
  • Databricks Runtime 14.3 LTS a vyšší na výpočetních prostředcích nakonfigurovaných s vyhrazeným režimem přístupu nebo bez režimu izolace.
• Přístup pro čtení k cestě kontrolního bodu, kterou používá streamovací dotaz.

Čtení stavového úložiště strukturovaného streamování

Číst můžete informace o úložišti stavu pro strukturované streamování dotazů spuštěné v jakémkoli podporovaném modulu Databricks Runtime. Použijte následující syntax:

Python

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')

Možnosti a schéma rozhraní API čtečky stavu

Úplný seznam statestore možností formátu najdete v tématu Úložiště stavů.

Výstupní data mají následující schéma:

Sloupec	Typ	Popis
key	Struktura (další typ odvozený ze stavového klíče)	Klíč pro záznam stavového operátoru v kontrolním bodu stavu.
value	Struktura (další typ odvozený z hodnoty stavu)	Hodnota pro záznam stavového operátoru v kontrolním bodě stavu.
partition_id	Celé číslo	Část stavového kontrolního bodu, která obsahuje záznam stavového operátoru.

V Databricks Runtime 16.4 LTS a vyšší, pokud readChangeFeed je možnost nastavena na true, výstupní data mají následující schéma:

Sloupec	Typ	Popis
batch_id	Dlouhý	ID dávky, k níž změna stavu náleží.
change_type	Řetězec	Typ změny použitý dávkou: update pro vložení a aktualizace, delete pro odstranění.
key	Struktura (další typ odvozený ze stavového klíče)	Klíč pro záznam stavového operátoru v kontrolním bodu stavu.
value	Struktura (další typ odvozený z hodnoty stavu)	Hodnota pro záznam stavového operátoru v kontrolním bodě stavu. null pro záznamy, kde change_type je delete.
partition_id	Celé číslo	Část stavového kontrolního bodu, která obsahuje záznam stavového operátoru.

Viz tabulkovou funkci read_statestore.

Čtení změn stavu strukturovaného streamování

K dispozici ve službě Databricks Runtime 16.4 LTS a vyšší. Chcete-li přečíst, jak se stav mění napříč mikrobatchy místo zobrazení celého stavu na jednom mikrobatchu, nastavte readChangeFeedtrue a určete changeStartBatchId. Volitelně můžete zadat changeEndBatchId. Úplný seznam možností najdete v tématu Úložiště stavů.

Například při čtení změn stavu od dávky 2 až po nejnovější potvrzenou dávku:

Python

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
);

Výstupní schéma obsahuje další batch_id a change_type sloupce. Úplné schéma najdete v tématu Možnosti a schéma rozhraní API čtečky stavu.

Čtení metadat stavu strukturovaného streamování

K dispozici ve službě Databricks Runtime 14.3 LTS nebo vyšší. Informace o metadatech stavu pro dotazy strukturovaného streamování můžete číst:

Python

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')

Vrácená data mají následující schéma:

Sloupec	Typ	Popis
operatorId	Celé číslo	Celočíselné ID stavového operátoru streamování.
operatorName	Řetězec	Název stavového streamovacího operátoru
stateStoreName	Řetězec	Název státního úložiště operátora.
numPartitions	Celé číslo	Počet oddílů stavového úložiště
minBatchId	Dlouhý	Minimální ID dávky, které je k dispozici pro dotazování stavu.
maxBatchId	Dlouhý	Maximální id dávky, které je k dispozici pro dotazování stavu.

Poznámka:

Hodnoty ID dávky dané minBatchId a maxBatchId odrážejí stav v okamžiku zápisu kontrolního bodu. Staré dávky se automaticky vyčistí pomocí mikrodávkového zpracování, takže poskytnutá hodnota není zaručeno, že bude stále dostupná.

Viz tabulkovou funkci read_state_metadata.

Příklad: Dotazování na jednu stranu spojení stream-stream

K dotazování na levou stranu spojení stream-stream použijte následující syntaxi:

Python

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'
);

Příklad: Úložiště stavu dotazu pro stream s několika stavovými operátory

V tomto příkladu se pomocí čtečky metadat stavu shromáždí podrobnosti o streamovacím dotazu s více stavovými operátory a výsledky metadat pak použije jako možnosti pro čtenáře stavu.

Čtenář metadat stavu vezme cestu kontrolního bodu jako jedinou možnost, jako v následujícím příkladu syntaxe:

Python

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')

Následující tabulka představuje příklad výstupu metadat úložiště stavů:

operatorId	jméno operátora	název úložiště stavu	početOddílů	minBatchId	maxBatchId
0	stateStoreSave	výchozí	200	0	13
1	dedupeWithinWatermark	výchozí	200	0	13

Pokud chcete získat výsledky pro dedupeWithinWatermark operátor, zadejte dotaz na čtenáře stavu s operatorId možností, jako v následujícím příkladu:

Python

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
);