Streamované čtení a zápisy tabulek Delta Lake

Tato stránka popisuje, jak používat tabulky Delta Lake jako zdroje a jímky pro strukturované streamování Sparku s readStream a writeStream. Delta Lake řeší běžné problémy s výkonem a spolehlivostí streamovaných systémů a souborů. Nabízí například tyto výhody:

• Sloučí malé soubory vytvořené při ingestování s nízkou latencí a zvýší výkonnost.
• Udržujte zpracování "přesně jednou" s více než jedním datovým proudem (nebo souběžnými dávkovými úlohami).
• Efektivní zjišťování nových souborů při použití souborů jako zdroje datového proudu

Informace o načtení dat pomocí streamovaných tabulek v Databricks SQL najdete v tématu Použití samostatných streamovacích tabulek.

Informace o statických spojeních datových proudů pomocí Delta Lake najdete v tématu Stream-static joins.

Úplný seznam možností DataStreamReader a DataStreamWriter pro Delta Lake viz DataStreamReader možnosti Delta Lake a DataStreamWriter možnosti Delta Lake.

Warning

Pokud jako zdroj streamování použijete tabulku Delta Lake, musí dotaz streamování běžet alespoň jednou v okně uchovávání zdrojové tabulky. Výchozí intervaly uchovávání informací jsou 7 dnů pro VACUUModebrané datové soubory a 30 dnů pro transakční protokol (logRetentionDuration). Pokud dotaz spadá za těmito okny, selže DELTA_FILE_NOT_FOUND_DETAILED a musí se resetovat pomocí úplné aktualizace.

Nenastavujte spark.sql.files.ignoreMissingFiles na true jako náhradní řešení, protože tato konfigurace bez upozornění vede k nesprávným výsledkům. Pokud plán datového proudu nestíhá výchozí dobu uchovávání, prodlužte místo toho dobu uchovávání zdrojové tabulky.

Použití tabulek Delta Lake jako jímky

Data můžete zapisovat do tabulky Delta Lake pomocí strukturovaného streamování. Transakční protokol Delta Lake zaručuje přesně jedno zpracování, i když jsou v tabulce spuštěné další streamy nebo dávkové dotazy.

Při zápisu do tabulky Delta Lake pomocí výstupu Structured Streaming se mohou objevit prázdné commity s epochId = -1. Jsou očekávané a obvykle k nim dochází:

• V první dávce každého spuštění streamovacího dotazu (k tomu dochází při každé dávce pro Trigger.AvailableNow).
• Při změně schématu (například přidání sloupce)

Tato prázdná commity jsou úmyslná a neoznačují chybu. Neovlivňují správnost ani výkon dotazu žádným významným způsobem.

Note

Funkce Delta Lake VACUUM odebere všechny soubory, které nespravuje Delta Lake, ale přeskočí všechny adresáře, které začínají _. Kontrolní body můžete bezpečně ukládat společně s dalšími daty a metadaty pro tabulku Delta Lake pomocí adresářové struktury, jako je například <table-name>/_checkpoints.

Monitorování backlogu s využitím metrik

K monitorování backlogu procesu dotazu streamování použijte následující metriky:

• numBytesOutstanding: Počet bajtů, které mají být ještě zpracovány ve frontě.
• numFilesOutstanding: Počet souborů, které ještě čekají na zpracování ve frontě.
• numNewListedFiles: Počet souborů Delta Lake uvedených pro výpočet backlogu pro tuto dávku.
• backlogEndOffset: Verze tabulky Delta Lake použitá k výpočtu backlogu.

V poznámkovém bloku si prohlédněte tyto metriky na kartě Nezpracovaná data na řídicím panelu průběhu dotazu streamování:

{
  "sources": [
    {
      "description": "DeltaSource[file:/path/to/source]",
      "metrics": {
        "numBytesOutstanding": "3456",
        "numFilesOutstanding": "8"
      }
    }
  ]
}

Režim připojování

Ve výchozím nastavení se streamy spouští v režimu připojení a přidávají do tabulky jenom nové záznamy.

Metodu toTable použijte při streamování do tabulek:

Python

(events.writeStream
   .outputMode("append")
   .option("checkpointLocation", "/tmp/delta/events/_checkpoints/")
   .toTable("events")
)

Scala

events.writeStream
  .outputMode("append")
  .option("checkpointLocation", "/tmp/delta/events/_checkpoints/")
  .toTable("events")

Režim dokončení

Použijte strukturované streamování v plném režimu k nahrazení celé tabulky po každé dávce. Agregovanou souhrnnou tabulku událostí můžete například průběžně aktualizovat podle zákazníka:

Python

(spark.readStream
  .table("events")
  .groupBy("customerId")
  .count()
  .writeStream
  .outputMode("complete")
  .option("checkpointLocation", "/tmp/delta/eventsByCustomer/_checkpoints/")
  .toTable("events_by_customer")
)

Scala

spark.readStream
  .table("events")
  .groupBy("customerId")
  .count()
  .writeStream
  .outputMode("complete")
  .option("checkpointLocation", "/tmp/delta/eventsByCustomer/_checkpoints/")
  .toTable("events_by_customer")

U aplikací bez striktních požadavků na latenci můžete ušetřit výpočetní prostředky a náklady pomocí jednorázových aktivačních událostí, jako je například AvailableNow. Pomocí tohoto triggeru můžete například aktualizovat souhrnné agregační tabulky podle daného plánu a zpracovat pouze nová data, která přišla od poslední aktualizace. Viz AvailableNow: Přírůstkové dávkové zpracování.

Zpracování změn zdrojových tabulek Delta Lake

Strukturované streamování přírůstkově čte tabulky Delta Lake. Když streamovací dotaz čte z tabulky Delta Lake, nové záznamy se zpracovávají idempotentně, jakmile jsou do zdrojové tabulky zapsány nové verze tabulky. Structured Streaming přijímá pouze vstupy v režimu append a vyvolá výjimku, pokud ve zdrojové tabulce Delta Lake dojde k jakýmkoli změnám. Například pokud operace UPDATE, DELETE, MERGE INTO nebo OVERWRITE upraví zdrojovou tabulku Delta Lake, kterou čte streamovací dotaz, stream selže s chybou.

Existují čtyři typické přístupy pro zpracování upstreamových změn zdrojových tabulek Delta Lake v závislosti na vašem případu použití. Referenční tabulka a podrobnosti o každé z nich jsou uvedené níže:

Přístup	Výhody	Nevýhody
skipChangeCommits	Jednoduché, nevyžaduje psaní složité logiky. Užitečné pro zpracování s pouze přidáváním, kde se změny v předchozím řetězci zpracovávají samostatně, nebo pro dočasné zpracování chybného záznamu.	Nepropaguje změny a zpracovává pouze přílohy.
Úplná aktualizace	Je to také jednoduché a nevyžaduje, abyste psali složitou logiku. Užitečné pro malé datové sady se vzácnými upstreamovými změnami.	Drahé pro velké datové sady. Vyžaduje opětovné zpracování všech podřízených tabulek.
Změna datového kanálu	Zpracovat všechny typy změn (vložení, aktualizace a odstranění) Databricks doporučuje streamování z CDC feedu tabulky Delta Lake namísto přímého streamování z tabulky, vždy když je to možné.	Vyžaduje, abyste napsali složitější logiku pro zpracování jednotlivých typů změn.
Materializované pohledy	Jednoduchá alternativa ke strukturovanému streamování, která má automatické šíření změn.	Vyšší latence. K dispozici pouze v pipelinech Lakeflow a v Databricks SQL.

Přeskočení upstreamových potvrzení změn pomocí skipChangeCommits

Nastavte skipChangeCommits na ignorování operací, které odstraňují nebo upravují existující záznamy, aby se zpracovávalo pouze přidávání. To je užitečné v případě, že změny existujících dat nemusí být šířené prostřednictvím datového proudu nebo pokud dáváte přednost samostatné logice pro zpracování těchto změn. Pokud potřebujete dočasně ignorovat jednorázové změny, můžete je zapnout a vypnout skipChangeCommits .

Databricks doporučuje používat skipChangeCommits pro většinu úloh, které nepoužívají kanály změn dat.

Python

(spark.readStream
  .option("skipChangeCommits", "true")
  .table("source_table")
)

Scala

spark.readStream
  .option("skipChangeCommits", "true")
  .table("source_table")

Important

Pokud se schéma tabulky Delta Lake změní po zahájení streamovaného čtení z tabulky, dotaz se nezdaří. U většiny změn schématu můžete stream restartovat, abyste vyřešili neshodu schématu a pokračovali ve zpracování.

V Databricks Runtime 12.2 LTS a starších verzích nemůžete streamovat data z tabulky Delta Lake s povoleným mapováním sloupců, která prošla neaditivní změnou schématu, například přejmenováním nebo odstraněním sloupců. Podrobnosti najdete v tématu Mapování sloupců a streamování.

Note

V Databricks Runtime 12.2 LTS a vyšší skipChangeCommits nahrazuje ignoreChanges. V Databricks Runtime 11.3 LTS a starších verzích je ignoreChanges jedinou podporovanou možností. Podrobnosti najdete v Možnost Legacy ignoreChanges.

Starší možnost: ignoreDeletes

ignoreDeletes je starší možnost, která zpracovává pouze transakce, které odstraňují data na hranicích oddílů (to znamená úplné odstranění oddílů). Pokud potřebujete zpracovat odstranění, aktualizace nebo jiné úpravy, které se netýkají oddílů, použijte místo toho skipChangeCommits.

Python

(spark.readStream
  .option("ignoreDeletes", "true")
  .table("user_events")
)

Scala

spark.readStream
  .option("ignoreDeletes", "true")
  .table("user_events")

Starší možnost: ignoreChanges

ignoreChanges je k dispozici v Databricks Runtime 11.3 LTS a starší. Ve verzi Databricks Runtime 12.2 LTS a vyšších je nahrazen skipChangeCommits.

S povolenou funkcí ignoreChanges se přepsané datové soubory ve zdrojové tabulce znovu emitují po operaci úpravy dat, jako je UPDATE, MERGE INTO, DELETE (v rámci oddílů) nebo OVERWRITE. Nezměněné řádky se často generují společně s novými řádky, takže podřízení příjemci musí být schopni zpracovávat duplicity. Smazání se nepropaguje do následujících procesů. ignoreChanges má přednost před ignoreDeletes.

Naproti tomu skipChangeCommits ignoruje operace změny souborů zcela. Přepsané datové soubory ve zdrojové tabulce kvůli operacím úprav dat, jako jsou UPDATE, MERGE INTO, DELETE a OVERWRITE, jsou zcela ignorovány. Pokud chcete odrážet změny ve zdrojových tabulkách datových proudů, musíte implementovat samostatnou logiku pro šíření těchto změn.

Databricks doporučuje používat skipChangeCommits pro všechny nové úlohy. Pokud chcete migrovat úlohu z ignoreChanges do skipChangeCommits, refaktorujte logiku streamování.

Úplná aktualizace podřízených tabulek

Pokud jsou upstreamové změny vzácné a data jsou dostatečně malá k opětovnému zpracování, můžete odstranit kontrolní bod streamování a výstupní tabulku a pak stream restartovat od začátku. To způsobí, že datový proud znovu zpracuje všechna data ze zdrojové tabulky. Mějte na paměti, že tento přístup také vyžaduje opětovné zpracování všech podřízených tabulek, které závisí na výstupu tohoto datového proudu.

Tento přístup je nejvhodnější pro menší datové sady nebo úlohy, u kterých jsou upstreamové změny zřídka časté a náklady na úplnou aktualizaci jsou přijatelné.

Použití informačního kanálu změn dat

Pro úlohy, které zpracovávají všechny typy změn (vložení, aktualizace a odstranění), použijte datový kanál změn Delta Lake. Datový kanál změn zaznamenává změny na úrovni řádků v tabulce Delta Lake, což umožňuje streamovat tyto změny a zapisovat logiku pro zpracování jednotlivých typů změn v podřízených tabulkách. Jedná se o nej robustnější přístup, protože váš kód explicitně zpracovává každý typ události změny. Viz Použití informačního kanálu změn dat v Azure Databricks.

Pokud používáte pipeliney Lakeflow, viz Rozhraní API AUTO CDC: Zjednodušte zachytávání změn dat pomocí pipelineů.

Important

V Databricks Runtime 12.2 LTS a starších verzích nemůžete streamovat data z kanálu change data feed pro tabulku Delta Lake s povoleným mapováním sloupců, která prošla neaditivní změnou schématu, například přejmenováním nebo odstraněním sloupců. Viz mapování sloupců a streamování.

Použití materializovaných zobrazení

Materializovaná zobrazení automaticky zpracovávají nadřazené změny opětovným dokončováním výsledků při změně zdrojových dat. Pokud nepotřebujete nejnižší možnou latenci a chcete se vyhnout správě složitosti streamování, může materializované zobrazení zjednodušit vaši architekturu. Materializovaná zobrazení jsou dostupná v kanálech Lakeflow a samostatných kanálech. Podívejte se na Materializovaná zobrazení.

Example

Předpokládejme například, že máte tabulku user_events s date, user_emaila action sloupce rozdělené podle date. Streamujete z tabulky user_events a potřebujete z ní odstranit data z důvodu GDPR.

skipChangeCommits umožňuje odstranit data v několika oddílech (v tomto příkladu filtrování user_email). Použijte následující syntax:

spark.readStream
  .option("skipChangeCommits", "true")
  .table("user_events")

Pokud aktualizujete user_email příkazem UPDATE, přepíše se soubor obsahující příslušný user_email. Slouží skipChangeCommits k ignorování změněně datových souborů.

Databricks doporučuje používat skipChangeCommits místo ignoreDeletes, pokud si nejste jisti, že při odstraněních dochází vždy k úplnému odstranění oddílů.

Použití foreachBatch pro zápisy do idempotentní tabulky

Note

Databricks doporučuje konfigurovat samostatný zápis streamování pro každou jímku, kterou chcete aktualizovat, místo použití foreachBatch. Zápisy do více jímek při foreachBatch redukci paralelizace a zvyšují celkovou latenci, protože zápisy do více tabulek jsou serializovány v foreachBatch.

Tabulky Delta Lake podporují následující DataFrameWriter možnosti, které umožňují učinit zápisy do více tabulek v rámci foreachBatch idempotentními:

• txnAppId: Jedinečný řetězec, který můžete předat při každém zápisu "DataFrame". Můžete například použít ID StreamingQuery jako txnAppId. txnAppId může být libovolný jedinečný řetězec vygenerovaný uživatelem a nemusí souviset s ID datového proudu.
• txnVersion: Monotonicky rostoucí číslo, které funguje jako verze transakce.

Delta Lake používá txnAppId a txnVersion identifikuje a ignoruje duplicitní zápisy. Například po selhání, které přerušilo dávkový zápis, můžete dávku znovu spustit se stejnými hodnotami txnAppId a txnVersion k tomu, aby správně identifikovaly a ignorovaly duplicity. Viz Použití příkazu foreachBatch k zápisu do libovolných datových jímek.

Warning

Pokud odstraníte kontrolní bod streamování a restartujete dotaz s novým kontrolním bodem, musíte zadat jiný txnAppId. Nové kontrolní body začínají ID dávky 0. Delta Lake používá ID dávky a txnAppId jako jedinečný klíč a přeskočí dávky s již viditelnými hodnotami.

Následující příklad kódu ukazuje tento vzor:

Python

app_id = ... # A unique string that is used as an application ID.

def writeToDeltaLakeTableIdempotent(batch_df, batch_id):
  batch_df.write.format(...).option("txnVersion", batch_id).option("txnAppId", app_id).save(...) # location 1
  batch_df.write.format(...).option("txnVersion", batch_id).option("txnAppId", app_id).save(...) # location 2

streamingDF.writeStream.foreachBatch(writeToDeltaLakeTableIdempotent).start()

Scala

val appId = ... // A unique string that is used as an application ID.
streamingDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
  batchDF.write.format(...).option("txnVersion", batchId).option("txnAppId", appId).save(...)  // location 1
  batchDF.write.format(...).option("txnVersion", batchId).option("txnAppId", appId).save(...)  // location 2
}

Upsertování ze streamovaných dotazů pomocí foreachBatch

Pomocí merge a foreachBatch můžete provádět složité upserty ze streamovacího dotazu do tabulky Delta Lake. Viz Použití příkazu foreachBatch k zápisu do libovolných datových jímek.

Tento přístup má mnoho aplikací:

• Zlepšení výkonu zápisu pomocí update výstupního režimu, zatímco complete výstupní režim vyžaduje přepis celé tabulky výsledků pro jednotlivé mikrobatchy.
• Nepřetržitě aplikujte stream změn v tabulce Delta Lake pomocí slučovacího dotazu k zápisu změn dat do foreachBatch. Viz. Pomalé změny dat (SCD) a zachytávání změn v datech (CDC) pomocí Delta Lake.
• Řešení deduplikace během zpracování proudu. K průběžnému zápisu dat do tabulky Delta Lake s automatickým odstraněním duplicitních dat můžete použít slučovací dotaz foreachBatch jen pro vložení. Podívejte se na odstranění duplicitních dat při zápisu do tabulek Delta Lake.

Note

• Ověřte, že příkaz merge uvnitř foreachBatch je idempotentní. V opačném případě mohou restartování streamovacích dotazů aplikovat operaci na stejnou dávku dat vícekrát. Viz též Použití foreachBatch pro idempotentní zápisy do tabulek.

• Pokud se merge používá v foreachBatch, metrika rychlosti přenosu vstupních dat může vrátit násobek skutečné rychlosti, jakou jsou data generována na zdroji. merge čte vstupní data několikrát, což vynásobí metriky. Aby se zabránilo násobení metrik, uložte datový rámec dávky merge do mezipaměti a poté ho po merge odstranění z mezipaměti uvolněte.

  Rychlost vstupních dat je k dispozici prostřednictvím StreamingQueryProgress a také v grafu rychlosti streamování v poznámkovém bloku. Viz dotazy Monitoring Structured Streaming na Azure Databricks.

Můžete například použít MERGE příkazy SQL v rámci foreachBatch:

Scala

// Function to upsert microBatchOutputDF into Delta Lake table using merge
def upsertToDelta(microBatchOutputDF: DataFrame, batchId: Long) {
  // Set the dataframe to view name
  microBatchOutputDF.createOrReplaceTempView("updates")

  // Use the view name to apply MERGE
  // NOTE: You have to use the SparkSession that has been used to define the `updates` dataframe
  microBatchOutputDF.sparkSession.sql(s"""
    MERGE INTO aggregates t
    USING updates s
    ON s.key = t.key
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
  """)
}

// Write the output of a streaming aggregation query into Delta Lake table
streamingAggregatesDF.writeStream
  .foreachBatch(upsertToDelta _)
  .outputMode("update")
  .start()

Python

# Function to upsert microBatchOutputDF into Delta Lake table using merge
def upsertToDelta(microBatchOutputDF, batchId):
  # Set the dataframe to view name
  microBatchOutputDF.createOrReplaceTempView("updates")

  # Use the view name to apply MERGE
  # NOTE: You have to use the SparkSession that has been used to define the `updates` dataframe

  # In Databricks Runtime 10.5 and below, you must use the following:
  # microBatchOutputDF._jdf.sparkSession().sql("""
  microBatchOutputDF.sparkSession.sql("""
    MERGE INTO aggregates t
    USING updates s
    ON s.key = t.key
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
  """)

# Write the output of a streaming aggregation query into Delta Lake table
(streamingAggregatesDF.writeStream
  .foreachBatch(upsertToDelta)
  .outputMode("update")
  .start()
)

Můžete také použít rozhraní Delta Lake API pro streamování upserts:

Scala

import io.delta.tables.*

val deltaTable = DeltaTable.forName(spark, "table_name")

// Function to upsert microBatchOutputDF into Delta Lake table using merge
def upsertToDelta(microBatchOutputDF: DataFrame, batchId: Long) {
  deltaTable.as("t")
    .merge(
      microBatchOutputDF.as("s"),
      "s.key = t.key")
    .whenMatched().updateAll()
    .whenNotMatched().insertAll()
    .execute()
}

// Write the output of a streaming aggregation query into Delta Lake table
streamingAggregatesDF.writeStream
  .foreachBatch(upsertToDelta _)
  .outputMode("update")
  .start()

Python

from delta.tables import *

deltaTable = DeltaTable.forName(spark, "table_name")

# Function to upsert microBatchOutputDF into Delta Lake table using merge
def upsertToDelta(microBatchOutputDF, batchId):
  (deltaTable.alias("t").merge(
      microBatchOutputDF.alias("s"),
      "s.key = t.key")
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute()
  )

# Write the output of a streaming aggregation query into Delta Lake table
(streamingAggregatesDF.writeStream
  .foreachBatch(upsertToDelta)
  .outputMode("update")
  .start()
)

Nastavení počáteční verze tabulky pro zpracování změn

Ve výchozím nastavení streamy začínají nejnovější dostupnou verzí tabulky Delta Lake. To zahrnuje kompletní snímek tabulky v tuto chvíli a všechny budoucí změny. Databricks doporučuje použít výchozí počáteční verzi tabulky pro většinu úloh.

Volitelně můžete pomocí následujících možností určit výchozí bod zdroje streamování Delta Lake bez zpracování celé tabulky.

• startingVersion: Verze tabulky Delta Lake, od které se má začít číst. Stream přečte všechny změny tabulky potvrzené v zadané verzi nebo po této verzi. Pokud zadaná verze není dostupná, stream se nespustí.

  Pokud chcete najít dostupné verze potvrzení, spusťte DESCRIBE HISTORY a zkontrolujte version. Chcete-li vrátit pouze nejnovější změny, zadejte latest. Informace o verzích tabulek Delta Lake najdete v tématu Práce s historií tabulek.

• startingTimestamp: Časové razítko, ze které chcete začít číst. Stream přečte všechny změny tabulky potvrzené v zadaném časovém razítku nebo po tomto časovém razítku. Pokud zadané časové razítko předchází všem potvrzením tabulky, streamované čtení začne s nejstarším dostupným časovým razítkem. Nastavte jednu z těchto nastavení:

  • Řetězec časového razítka. Například "2019-01-01T00:00:00.000Z".
  • Řetězec kalendářního data. Například "2019-01-01".

Nemůžete nastavit obě startingVersion a startingTimestamp současně. Tato nastavení platí jenom pro nové dotazy streamování. Pokud se spustil dotaz na streamování a průběh byl zaznamenán v jeho kontrolním bodu, tato nastavení se ignorují.

Important

I když zdroj streamování můžete spustit ze zadané verze nebo časového razítka, schéma zdroje streamování je vždy nejnovější schéma tabulky Delta Lake. Po zadané verzi nebo časovém razítku je nutné zajistit, aby nedošlo k žádné nekompatibilní změně schématu tabulky Delta Lake. Jinak může zdroj streamování vrátit nesprávné výsledky při čtení dat s nesprávným schématem.

Example

Předpokládejme například, že máte tabulku user_events. Pokud chcete číst změny od verze 5, použijte:

spark.readStream
  .option("startingVersion", "5")
  .table("user_events")

Pokud chcete číst změny od 10. 10. 2018, použijte:

spark.readStream
  .option("startingTimestamp", "2018-10-18")
  .table("user_events")

Zpracování počátečního snímku bez vyřazení dat

Tato funkce je dostupná pro Databricks Runtime 11.3 LTS a vyšší.

Ve stavovém streamovacím dotazu s definovaným vodoznakem může zpracování souborů podle času úprav zpracovávat záznamy v nesprávném pořadí. To může způsobit, že vodoznak nesprávně označí záznamy jako zpožděné události a zahodí je. K tomu může dojít pouze v případě, že se počáteční snímek Delta zpracuje ve výchozím pořadí.

U datových proudů se zdrojovou tabulkou Delta nejprve dotaz zpracuje všechna data, která jsou v tabulce, a vytvoří verzi s názvem počáteční snímek. Ve výchozím nastavení se datové soubory tabulky Delta Lake zpracovávají na základě toho, který soubor byl naposledy změněn. Čas poslední změny však nemusí nutně představovat pořadí času události záznamu.

Pokud se chcete vyhnout poklesům dat během počátečního zpracování snímků, povolte možnost withEventTimeOrder . withEventTimeOrder rozdělí časový rozsah událostí počátečních dat snímků do časových intervalů. Každý mikrodávk zpracovává kbelík filtrováním dat v rámci časového rozsahu. Možnosti maxFilesPerTrigger a maxBytesPerTrigger se stále vztahují k řízení velikosti mikrodávky, ale pouze přibližně z důvodu způsobu zpracování.

Následující diagram znázorňuje tento proces:

Omezení

• Pokud se dotaz streamu spustil a počáteční snímek se aktivně zpracovává, nemůžete ho změnit withEventTimeOrder . Pokud chcete restartovat s withEventTimeOrder změněným, musíte kontrolní bod odstranit.
• Pokud withEventTimeOrder je tato funkce povolená, nemůžete stream downgradovat na verzi Databricks Runtime, která tuto funkci nepodporuje, dokud se počáteční zpracování snímků nedokončí. Pokud chcete downgradovat, počkejte na dokončení počátečního snímku nebo odstraňte kontrolní bod a restartujte dotaz.
• Tato funkce není podporována v následujících scénářích:
  • Sloupec času události je generovaný sloupec a mezi zdrojem Delta a vodoznakem existují neprojekční transformace.
  • V dotazu streamu je vodoznak, který obsahuje více než jeden zdroj Delta.

výkon

Pokud withEventTimeOrder je povoleno, může být výkon počátečního zpracování snímků pomalejší. Každá mikrodávka prohledá počáteční snímek a vyfiltruje data v odpovídajícím časovém rozsahu událostí. Zvýšení výkonu filtrování:

• Jako čas události použijte zdrojový sloupec Delta, aby bylo možné použít přeskočení dat. Viz Přeskočení dat.
• Rozdělte tabulku podle sloupce času události.

Pomocí uživatelského rozhraní Sparku zjistíte, kolik souborů Delta se kontroluje pro konkrétní mikrodávku.

Example

Předpokládejme, že máte tabulku user_events se sloupcem event_time. Dotaz streamování je agregační dotaz. Pokud chcete zajistit, aby během počátečního zpracování snímků nedošlo k žádnému poklesu dat, můžete použít:

spark.readStream
  .option("withEventTimeOrder", "true")
  .table("user_events")
  .withWatermark("event_time", "10 seconds")

Můžete nastavit withEventTimeOrder konfiguraci Sparku v clusteru, aby se použila pro všechny dotazy streamování: spark.databricks.delta.withEventTimeOrder.enabled true.

Omezení vstupní rychlosti za účelem zlepšení výkonu zpracování

Ve výchozím nastavení zpracovává strukturované streamování co nejvíce souborů v každé mikrodávce. Pokud chcete omezit množství zpracovávaných dat na dávku a spravovat využití paměti, stabilizovat latenci nebo snížit náklady na cloudové úložiště, použijte následující možnosti:

• maxFilesPerTrigger: Počet nových souborů, které mají být zváženy v každé mikro šarži. Výchozí hodnota je 1 000.
• maxBytesPerTrigger: Množství dat, která se zpracovávají v každé mikrodávce. Tato možnost nastaví „měkký limit“, což znamená, že dávkové zpracování zpracovává přibližně toto množství dat a může zpracovat více než limit, aby se streamovací dotaz mohl posunout vpřed v případech, kdy je nejmenší vstupní jednotka větší než tento limit. Tato možnost není ve výchozím nastavení nastavená.

Pokud použijete jak maxBytesPerTrigger a maxFilesPerTrigger, mikrodávkový proces zpracovává data, dokud není dosaženo buď maxFilesPerTrigger nebo maxBytesPerTrigger limitu.

Note

Pokud ve výchozím nastavení logRetentionDuration vyčistí transakce ve zdrojové tabulce a dotaz streamování se pokusí zpracovat tyto verze, dotaz se nezdaří zabránit ztrátě dat. Možnost failOnDataLoss můžete nastavit tak, aby false ignorovala ztracená data a pokračovala ve zpracování. Viz Konfigurace uchovávání dat pro časové dotazy.

Řízení nákladů na cloudové úložiště

Dotazy streamování mají k dispozici několik spouštěcích režimů, které umožňují vyrovnávat náklady a latenci, včetně processingTime, availableNow a realTime. Viz Řízení nákladů na cloudové úložiště.