foreach_sink

Dekorátor @dp.foreach_sink() definuje sink, ForEach který zpracovává proud po řádcích pomocí vlastní Python logiky. Odkazujete na sink jako na target v append flow nebo update flow pro zápis dat. Pro konceptuální vedení, úvahy a příklady viz Použijte ForEach dřez k zápisu každého řádku v pipeline. Pro zpracování každého mikro-dávkování jako DataFrame viz foreach_batch_sink.

Syntax

Pro definici ForEach dřezu z funkce ozdobte funkci :@dp.foreach_sink()

from pyspark import pipelines as dp

@dp.foreach_sink(name="<name>")
def row_handler(row):
    """
    Required:
      - `row`: a Row that contains one record of the stream.
    """
    # Your custom write logic here.
    # This function runs on the executors, so import any libraries it uses inside the function.

Pro definici spotřebiče ForEach z writer objektu zavolejte a předejte dp.foreach_sink() objekt s argumentem writer :

from pyspark import pipelines as dp

class RowWriter:
    def open(self, partition_id, epoch_id):
        # Optional. Return False to skip the rows of this partition.
        return True

    def process(self, row):
        # Required. Called for each row.
        ...

    def close(self, error):
        # Optional. `error` is None if processing succeeded.
        ...

dp.foreach_sink(name="<name>", writer=RowWriter())

Můžete také předat funkci do s dp.foreach_sink() argumentem func místo použití dekorátoru, například dp.foreach_sink(name="<name>", func=row_handler).

Parameters

Parameter	Description
řádku	Required. A Row , která obsahuje jeden záznam o proudu. Vaše funkce, nebo process metoda vašeho autora, dostane jeden řádek pro každý hovor.
name	Optional. Jedinečný název pro identifikaci jímky v kanálu. Výchozí je název funkce. Pro writer objekt se výchozí název třídy používá malými písmeny.
func	Optional. Funkce, kterou pipeline volá pro každý řádek. Používejte func to při přímém volání dp.foreach_sink() , ne jako dekoratér. Nemůžete projít obou, func a writer.
writer	Optional. Objekt, který implementuje metodu process(row) , a volitelně open(partition_id, epoch_id) i metody and close(error) . Předložte to jako argument na základě klíčových slov. Nemůžete projít obou, func a writer.

Metody spisovatelů

Writer objekt používá stejné metody jako Apache Spark Structured Streaming foreach writer. Pro každou partition každé mikrodávky pipeline volá open, pak volá process pro každý řádek, pokud open vrací True, a poté volá close. Pro informace o psaní vlastních výstupů ve strukturovaném streamování viz Použít foreachBatch pro zápis do libovolných datových spotřebičů.

Metoda	Description
open(partition_id, epoch_id)	Optional. Pipeline volá tuto metodu dříve, než zpracuje řádky partition pro mikrodávku. Používejte ho k navazování spojení nebo zahájení transakcí. Vraťte True se pro zpracování řádků partitionu nebo False pro jejich přeskočte. Pokud zapisovatel nedefinuje open, pipeline zpracovává řádky každé partition. partition_id a epoch_id může se opakovat napříč toky, které zapisují do stejného spotřebiče a po plném obnovení. Než je použijete k deduplikaci řádků, podívejte se na Řešení neúspěchů a duplikace řádků.
process(row)	Required. Pipeline volá tuto metodu pro každý řádek.
close(error)	Optional. Pipeline volá tuto metodu poté, co zpracuje řádky partition. error je výjimka, která nastala při zpracování řádků, nebo None pokud zpracování bylo úspěšné. Použijte ho k uzavření kontaktů. Pokud open jste transakci spustili, potvrďte ji pouze tehdy, když error je , Nonea jinak ji vraťte zpět.