foreach_batch_sink

Dekorátor @dp.foreach_batch_sink() definuje jímku ForEachBatch, která zpracovává datový proud jako řadu mikrodávek, které zpracováváte v Python s vlastní logikou. Na jímku odkazujete jako target na tok připojení k zápisu transformovaných dat. Koncepční pokyny, důležité informace a příklady najdete v tématu Použití příkazu ForEachBatch k zápisu do libovolných datových jímek v kanálech.

Syntax

from pyspark import pipelines as dp

@dp.foreach_batch_sink(name="<name>")
def batch_handler(df, batch_id):
    """
    Required:
      - `df`: a Spark DataFrame representing the rows of this micro-batch.
      - `batch_id`: unique integer ID for each micro-batch in the query.
    """
    # Your custom write or transformation logic here
    # Example:
    # df.write.format("some-target-system").save("...")
    #
    # To access the sparkSession inside the batch handler, use df.sparkSession.

Parameters

Parametr	Description
name	Optional. Jedinečný název pro identifikaci jímky v kanálu. Výchozí hodnota je název uživatelsky definované funkce, pokud není zahrnut.
batch_handler	Toto je uživatelem definovaná funkce (UDF), která se volá pro každou mikrodávku.
Df	Datový rámec Sparku obsahující data pro aktuální mikrodávku
batch_id	Celočíselné ID mikrodávky. Spark zvýší toto ID pro každý interval triggeru. batch_id A 0 představuje začátek datového proudu nebo začátek úplné aktualizace. Kód foreach_batch_sink by měl správně zpracovat úplnou aktualizaci podřízených zdrojů dat. Další informace naleznete v tématu Úplná aktualizace.