foreach (DataStreamWriter)

Nastaví výstup streamovacího dotazu, který se má zpracovat pomocí poskytnutého zapisovače. Logiku zpracování je možné zadat jako funkci, která přebírá řádek jako vstup, nebo jako objekt s volitelnými process(row) a open(partition_id, epoch_id) metodamiclose(error).

Syntaxe

foreach(f)

Parametry

Parameter	Typ	Description
f	volatelný nebo objekt	Funkce, která přebírá jako vstup řádek nebo objekt s metodou process(row) a volitelnými open metodami a close metodami.

Návraty

DataStreamWriter

Poznámky

Zadaný objekt musí být serializovatelný. Jakákoli inicializace pro zápis dat (například otevření připojení) by měla být provedena uvnitř open(), ne v době výstavby.

Příklady

import time
df = spark.readStream.format("rate").load()

Zpracování každého řádku pomocí funkce:

def print_row(row):
    print(row)

q = df.writeStream.foreach(print_row).start()
time.sleep(3)
q.stop()

Zpracovat každý řádek pomocí objektu s open, processa close metody:

class RowPrinter:
    def open(self, partition_id, epoch_id):
        print("Opened %d, %d" % (partition_id, epoch_id))
        return True

    def process(self, row):
        print(row)

    def close(self, error):
        print("Closed with error: %s" % str(error))

q = df.writeStream.foreach(RowPrinter()).start()
time.sleep(3)
q.stop()