foreach_sink

A @dp.foreach_sink() dekorátor egy ForEach sink-et határoz meg, amely egy soronként dolgozza fel az áramlatot egyedi Python logikával. Az elnyelőt targetappend flow-ban vagy frissítési folyamatként hivatkozod az adatok megírásához. A koncepcionális útmutatásért, megfontolásért és példákért lásd: Használj egy ForEach sinket a pipelines sorainak megírásához. Ha minden mikro-adag adatkeretként dolgozza fel, lásd foreach_batch_sink.

Syntax

A ForEach mosogató meghatározásához egy függvényből díszítsük a függvényt a következőkkel @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.

Egy író objektumból történő ForEach sink definiálásához hívjuk dp.foreach_sink() és továbbítsuk az objektumot az argumentummal 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())

A dp.foreach_sink() függvényt is továbbíthatod az func argumentummal a dekorátor helyett, például dp.foreach_sink(name="<name>", func=row_handler).

Parameters

Paraméter	Description
sor	Required. A, Row amely tartalmaz egy adatfolyam felvételét. A te funkciód, vagyis process az íród módszere, minden híváshoz egy sort kap.
name	Optional. Egyedi név a csatornán belüli nyelő azonosításához. Alapértelmezés a függvény nevére van jellemző. Egy író objektumnál alapértelmezés szerint kisbetűs osztálynévre van írva.
func	Optional. Az a függvény, amit a csővezeték minden sorra hív. Használd func , amikor közvetlenül hívsz dp.foreach_sink() , nem dekorátorként. Nem tudod átmenni mindkettőt func , és writer.
writer	Optional. Egy objektum, amely megvalósítja a process(row) metódust, és opcionálisan az open(partition_id, epoch_id) és close(error) metódusokat. Add át kulcsszavaként az érvet. Nem tudod átmenni mindkettőt func , és writer.

Írói módszerek

Egy írói objektum ugyanazokat a módszereket használja, mint az Apache Spark Structured Streaming foreach író. Minden mikro-tétel partíciójához a csővezeték hívja open, majd minden sort hív process , ha open visszaadja True, és aztán hívja close. A Strukturált Streamingben egyedi kimenetek írásáról információért lásd: Use foreachBatch to write to write to random data sink.

Method	Description
open(partition_id, epoch_id)	Optional. A csővezeték ezt a módszert hívja, mielőtt feldolgozná a partíció sorait egy mikro-batch esetében. Használd kapcsolatok megnyitására vagy tranzakciók indítására. Return True a partíció sorainak feldolgozásához vagy False kihagyásához. Ha az író nem definiálja open, a csővezeték feldolgozza az egyes partíciók sorait. partition_id és epoch_id ismétlődhet olyan folyamatokon is, amelyek ugyanahhoz a vízgyűjtőhöz írnak, teljes frissítés után. Mielőtt ezeket használnád sorok deduplikálására, lásd a hibák kezelését és a sorok duplikálását.
process(row)	Required. A csővezeték minden sorra ezt a módszert hívja.
close(error)	Optional. A csővezeték ezt a módszert a partíció sorai feldolgozása után hívja meg. error ez a kivétel, amely sorfeldolgozás közben történt, vagy None ha a feldolgozás sikeres volt. Használd a kapcsolatok lezárására. Ha open tranzakciót indítanak, csak akkor kötelezd fel, ha error van None, és akkor vissza a visszajelzést.