Použití příkazu ForEachBatch k zápisu do libovolných datových jímek v kanálech

ForEachBatch sink zpracovává stream jako řadu mikrodávek. Každou dávku je možné zpracovat v Pythonu s vlastní logikou podobnou strukturovanému streamování Apache Spark foreachBatch. S deklarativními kanály Sparku Lakeflow (SDP) ForEachBatch můžete transformovat, slučovat nebo zapisovat streamovaná data do jednoho nebo více cílů, které nativně nepodporují zápisy streamovaných dat.

Jímka ForEachBatch poskytuje následující funkce:

  • Vlastní logika pro každou mikrodávku: ForEachBatch je flexibilní výstup pro streamování. Pomocí kódu Python můžete použít libovolné akce (například sloučení do externí tabulky, zápis do více cílů nebo provádění upsertů).
  • Podpora úplné aktualizace: Kanály spravují kontrolní body podle toku, takže se kontrolní body automaticky resetují, když provedete úplnou aktualizaci kanálu. S použitím ForEachBatch sink jste zodpovědní za správu resetování navazujících dat, když k tomu dojde.
  • Podpora katalogu Unity: Jímka ForEachBatch podporuje všechny funkce katalogu Unity, jako je čtení nebo zápis do svazků nebo tabulek katalogu Unity.
  • Omezená správa zdrojů: Automatizovaný proces nesleduje, jaká data se zapisují do úložiště ForEachBatch, takže tato data nelze vyčistit. Zodpovídáte za veškerou správu podřízených dat.
  • Položky protokolu událostí: Protokol událostí kanálu zaznamenává vytvoření a použití každé jímky ForEachBatch. Pokud vaše Python funkce není serializovatelná, zobrazí se v protokolu událostí upozornění s dalšími návrhy.

Poznámka:

  • Jímka ForEachBatch je určená pro dotazy streamování, například append_flow. Není určen pro pipelines určené pouze pro dávkové zpracování nebo pro AutoCDC sémantiku.
  • Jímka ForEachBatch popsaná na této stránce je určená pro kanály. Strukturované streamování Apache Sparku také podporuje foreachBatch. Informace o strukturovaném streamování foreachBatchnaleznete v tématu Použití příkazu foreachBatch k zápisu do libovolných datových jímek.

Kdy použít jímku ForEachBatch

Použijte jímku ForEachBatch pokaždé, když kanál vyžaduje funkce, které nejsou k dispozici prostřednictvím integrovaného formátu jímky, například deltanebo kafka. Mezi obvyklé případy použití patří:

  • Sloučení nebo upsertování do tabulky Delta Lake: Spusťte vlastní logiku sloučení pro každou mikrodávku (například zpracování aktualizovaných záznamů).
  • Zápis do více nebo nepodporovaných cílů: Zapište výstup každé dávky do více tabulek nebo externích úložných systémů, které nepodporují streamované zápisy (například určité zdroje JDBC).
  • Aplyování vlastní logiky nebo transformací: Přímá manipulace s daty v Python (například pomocí specializovaných knihoven nebo pokročilých transformací).

Informace o předdefinovaných jímkách nebo vytváření vlastních jímek pomocí Python najdete v tématu Sinks in Lakeflow Spark Declarative Pipelines.

Referenční informace @dp.foreach_batch_sink() k rozhraní API pro Python najdete v foreach_batch_sink.

Úplná aktualizace

Protože ForEachBatch používá streamovací dotaz, kanál sleduje adresář kontrolních bodů pro každý tok. Při úplné aktualizaci:

  • Adresář kontrolních bodů se resetuje.
  • Funkce odtoku (foreach_batch_sink UDF) zaznamenává úplně nový batch_id cyklus počínaje od 0.
  • Data v cílovém systému se kanálem automaticky nevyčistí (protože kanál neví, kde se zapisují vaše data). Pokud potřebujete scénář čisté slate, musíte ručně odstranit nebo zkrátit externí tabulky nebo umístění, které vaše jímka ForEachBatch naplní.

Použití funkcí katalogu Unity

Všechny stávající funkce katalogu Unity v Apache Spark Structured Streaming foreach_batch_sink zůstanou dostupné.

To zahrnuje zápis do spravovaných nebo externích tabulek katalogu Unity. Mikrodávky můžete zapisovat do spravovaných nebo externích tabulek Unity Catalog, přesně tak jako v jakékoli úloze strukturovaného streamování Apache Spark.

Položky protokolu událostí

Když vytvoříte jímku ForEachBatch, událost SinkDefinition s "format": "foreachBatch" je přidána do protokolu událostí zpracování dat.

Díky tomu můžete sledovat využití jímek ForEachBatch a zobrazit upozornění týkající se jímky.

Použití s Databricks Connect

Pokud zadáváte funkci , která není serializovatelná (důležitým požadavkem pro Databricks Connect), obsahuje WARN protokol událostí záznam, který doporučuje zjednodušit nebo refaktorovat kód, pokud je vyžadována podpora Databricks Connect.

Pokud například použijete dbutils k získání parametrů v rámci funkce ForEachBatch UDF, můžete místo toho získat argument před použitím v UDF:

# Instead of accessing parameters within the UDF...
def foreach_batch(df, batchId):
  value = dbutils.widgets.get ("X") + str (i)

# ...get the parameters first, and use them within the UDF:
argX = dbutils.widgets.get ("X")

def foreach_batch(df, batchId):
  value = argX + str (i)

Osvědčené postupy

  1. Udržujte funkci ForEachBatch stručnou: Vyhněte se práci s vlákny, náročným závislostem knihoven nebo manipulaci s velkými daty v paměti. Složitá nebo stavová logika může vést k chybám serializace nebo kritickým bodům výkonu.
  2. Monitorujte složku kontrolního bodu: U dotazů streamování spravuje protokol SDP kontrolní body podle toku, nikoli jímky. Pokud máte v kanálu více toků, každý tok má svůj vlastní adresář kontrolních bodů.
  3. Ověření externích závislostí: Pokud spoléháte na externí systémy nebo knihovny, zkontrolujte, jestli jsou nainstalované na všech uzlech clusteru nebo ve vašem kontejneru.
  4. Mějte na paměti Databricks Connect: Pokud se vaše prostředí může v budoucnu přesunout na Databricks Connect, zkontrolujte, jestli je váš kód serializovatelný a nespoléhá na dbutils v rámci foreach_batch_sink UDF.

Omezení

  • Žádné údržby pro ForEachBatch: Protože váš vlastní kód v Pythonu může zapisovat data kamkoli, proces nemůže tato data vyčistit ani sledovat. Pro cíle, do které zapisujete, musíte zpracovávat vlastní zásady správy dat nebo uchovávání informací.
  • Metriky v mikrodávce: Kanály shromažďují metriky streamování, ale některé scénáře můžou při použití ForEachBatch způsobit neúplné nebo neobvyklé metriky. Důvodem je základní flexibilita forEachBatch, která ztěžuje sledování toků dat a řádků pro systém.
  • Podpora zápisu do více cílů bez více čtení: Někteří zákazníci mohou použít ForEachBatch ke čtení ze zdroje jednou a pak zapisovat do více cílů. Abyste toho dosáhli, musíte zahrnout df.persist nebo df.cache do funkce ForEachBatch. Pomocí těchto možností se Azure Databricks pokusí číst data jen jednou. Bez těchto možností povede váš dotaz k více čtením. Tento kód není součástí následujících příkladů kódu.
  • Použití se službou Databricks Connect: Pokud vaše pipeline běží na Databricks Connect, foreachBatch uživatelské definované funkce (UDF) musí být serializovatelné a nelze je použít dbutils. Pipeline vyvolá upozornění, pokud zjistí neserializovatelnou uživatelsky definovanou funkci, ale pipeline neselže.
  • Logika, která není serializovatelná: Kód odkazující na místní objekty, třídy nebo nerozbalitelné prostředky může narušit fungování v kontextech Databricks Connect. Používejte čistě Python moduly a ověřte, že odkazy (například dbutils) se nepoužívají, pokud je požadavkem Databricks Connect.

Examples

Příklad základní syntaxe

from pyspark import pipelines as dp

# Create a ForEachBatch sink
@dp.foreach_batch_sink(name = "my_foreachbatch_sink")
def feb_sink(df, batch_id):
  # Custom logic here. You can perform merges,
  # write to multiple destinations, etc.
  return

# Create source data for example:
@dp.table()
def example_source_data():
  return spark.range(5)

# Add sink to an append flow:
@dp.append_flow(
    target="my_foreachbatch_sink",
)
def my_flow():
  return spark.readStream.format("delta").table("example_source_data")

Použití ukázkových dat pro jednoduché potrubí

V tomto příkladu se používá ukázka taxislužby NYC. Předpokládá se, že správce pracovního prostoru povolil katalog Databricks Public Datasets. V případě jímky přejděte my_catalog.my_schema do katalogu a schématu, ke kterému máte přístup.

from pyspark import pipelines as dp
from pyspark.sql.functions import current_timestamp

# Create foreachBatch sink
@dp.foreach_batch_sink(name = "my_foreach_sink")
def my_foreach_sink(df, batch_id):
    # Custom logic here. You can perform merges,
    # write to multiple destinations, etc.
    # For this example, we are adding a timestamp column.
    enriched = df.withColumn("processed_timestamp", current_timestamp())
    # Write to a Delta location
    enriched.write \
      .format("delta") \
      .mode("append") \
      .saveAsTable("my_catalog.my_schema.trips_sink_delta")
    # Return is optional here, but generally not used for the sink
    return

# Create an append flow that reads sample data,
# and sends it to the ForEachBatch sink
@dp.append_flow(
    target="my_foreach_sink",
)
def taxi_source():
  df = spark.readStream.table("samples.nyctaxi.trips")
  return df

Zápis do více cílových míst

Tento příklad zapisuje do více cílů. Ukazuje použití txnVersion a txnAppId, aby zápisy do tabulek Delta Lake byly idempotentní. Podrobnosti najdete v tématu Použijte foreachBatch pro idempotentní zápisy tabulek.

Předpokládejme, že píšeme do dvou tabulek, table_a a table_b, a předpokládejme, že v dávce bude zápis do table_a úspěšný, zatímco zápis do table_b selže. Když se dávka znovu spustí, pár (txnVersion, txnAppId) umožní Delta ignorovat duplicitní zápis do table_aa zapsat pouze dávku do table_b.

from pyspark import pipelines as dp

app_id = "my-app-name" # different applications that write to the same table should have unique txnAppId

# Create the ForEachBatch sink
@dp.foreach_batch_sink(name="user_events_feb")
def user_events_handler(df, batch_id):
    # Optionally do transformations, logging, or merging logic
    # ...

    # Write to a Delta table
    df.write \
     .format("delta") \
     .mode("append") \
     .option("txnVersion", batch_id) \
     .option("txnAppId", app_id) \
     .saveAsTable("my_catalog.my_schema.example_table_1")

    # Also write to a JSON file location
    df.write \
      .format("json") \
      .mode("append") \
      .option("txnVersion", batch_id) \
      .option("txnAppId", app_id) \
      .save("/tmp/json_target")
    return

# Create source data for example
@dp.table()
def example_source():
  return spark.range(5)


# Create the append flow, and target the ForEachBatch sink
@dp.append_flow(target="user_events_feb", name="user_events_flow")
def read_user_events():
    return spark.readStream.format("delta").table("example_source")

Použití spark.sql()

V jímce ForEachBatch můžete použít spark.sql() jako v následujícím příkladu.

from pyspark import pipelines as dp
from pyspark.sql import Row

@dp.foreach_batch_sink(name = "example_sink")
def feb_sink(df, batch_id):
  df.createOrReplaceTempView("df_view")
  df.sparkSession.sql("MERGE INTO target_table AS tgt " +
            "USING df_view AS src ON tgt.id = src.id " +
            "WHEN MATCHED THEN UPDATE SET tgt.id = src.id * 10 " +
            "WHEN NOT MATCHED THEN INSERT (id) VALUES (id)"
          )
  return

# Create target delta table
spark.range(5).write.format("delta").mode("overwrite").saveAsTable("target_table")

# Create source table
@dp.table()
def src_table():
  return spark.range(5)

@dp.append_flow(
    target="example_sink",
)
def example_flow():
  return spark.readStream.format("delta").table("source_table")

Sloučení s externí tabulkou Delta Lake

from pyspark import pipelines as dp
from pyspark.sql.functions import col
from delta.tables import DeltaTable

@dp.foreach_batch_sink(name = "external_merge_feb")
def foreachBatchFunc(df, batchId):
  out = DeltaTable.forName(df.sparkSession, $table)
  out.alias("target") \
    .merge(df.alias("source"), "source.value = target.value") \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .whenNotMatchedBySourceDelete() \
    .execute()

@dp.update_flow(
    target="external_merge_feb",
    name="merge_flow"
)
def read_data():
    return (
        spark.readStream.format("delta")
        .load("/tmp/source_delta_table")
        .filter(col("value").isNotNull())
    )

Často kladené otázky (FAQ)

Můžu použít dbutils v jímce ForEachBatch?

Pokud plánujete pipeline spustit v prostředí mimo Databricks Connect, dbutils může fungovat. Pokud ale používáte Databricks Connect, dbutils není ve vaší foreachBatch funkci přístupný. Potrubí může vyvolat upozornění, pokud zjistí dbutils použití, abyste se vyhnuli problémům.

Můžu použít více toků s jednou jímkou ForEachBatch?

Ano. Můžete definovat více toků (se @dp.append_flow), které cílují na stejný název jímky, ale každý z nich udržuje vlastní kontrolní body.

Zpracovává potrubí uchovávání nebo čištění dat pro můj cílový systém?

Ne. Protože jímka ForEachBatch může zapisovat do libovolného umístění nebo systému, kanál nemůže automaticky spravovat ani odstraňovat data v daném cíli. Tyto operace musíte zpracovat jako součást vlastního kódu nebo externích procesů.

Jak můžu řešit chyby serializace nebo chyby ve funkci ForEachBatch?

Prohlédněte si protokoly ovladačů clusteru nebo protokoly událostí pipeline. V případě problémů se serializací souvisejících se Spark Connect zkontrolujte, jestli vaše funkce závisí pouze na serializovatelných Python objektech a neodkazuje na nepovolené objekty (například otevřené popisovače souborů nebo dbutils).

  • Last updated on