DataSourceStreamArrowWriter

Základní třída pro zapisovače datových proudů, které zpracovávají data pomocí PyArrow's RecordBatch.

Na rozdíl od DataSourceStreamWritertoho, který funguje s iterátorem objektů Spark Row , je tato třída optimalizována pro formát Šipka při zápisu streamovaných dat. Může nabídnout lepší výkon při spolupráci se systémy nebo knihovnami, které nativně podporují šipku pro případy použití streamování. Implementujte tuto třídu a vraťte instanci, DataSource.streamWriter() aby byl zdroj dat zapisovatelný jako jímka streamování pomocí šipky.

Syntaxe

from pyspark.sql.datasource import DataSourceStreamArrowWriter

class MyDataSourceStreamArrowWriter(DataSourceStreamArrowWriter):
    def write(self, iterator):
        ...

Methods

Metoda	Description
write(iterator)	Zapíše iterátor objektů PyArrow RecordBatch do jímky streamování. Volali jsme exekutory jednou za mikrobatch. WriterCommitMessageVrátí zprávu , nebo None pokud neexistuje žádná potvrzovací zpráva. Tato metoda je abstraktní a musí být implementována.
commit(messages, batchId)	Potvrdí mikrobatch pomocí seznamu zpráv potvrzení shromážděných ze všech exekutorů. Vyvolá se na ovladači, když se všechny úlohy v mikrobatchu úspěšně spustí. Zděděno z DataSourceStreamWriter.
abort(messages, batchId)	Přeruší mikrobatch pomocí seznamu zpráv potvrzení shromážděných ze všech exekutorů. Vyvoláno na ovladači, když došlo k selhání jedné nebo více úloh v mikrobatchu. Zděděno z DataSourceStreamWriter.

Poznámky

• Ovladač shromažďuje zprávy potvrzení ze všech exekutorů a předává je, commit() pokud jsou všechny úkoly úspěšné nebo pokud abort() některý úkol selže.
• Pokud úloha zápisu selže, zpráva potvrzení bude None v seznamu předána commit() nebo abort().
• batchId jednoznačně identifikuje jednotlivé mikrobatchy a přírůstky o 1 s každým zpracovaným mikrobatchem.

Příklady

Implementujte zapisovač datových proudů založený na šipkách, který počítá řádky na mikrobatch:

from dataclasses import dataclass
from pyspark.sql.datasource import DataSource, DataSourceStreamArrowWriter, WriterCommitMessage

@dataclass
class MyCommitMessage(WriterCommitMessage):
    num_rows: int

class MyDataSourceStreamArrowWriter(DataSourceStreamArrowWriter):
    def write(self, iterator):
        total_rows = 0
        for batch in iterator:
            total_rows += len(batch)
        return MyCommitMessage(num_rows=total_rows)

    def commit(self, messages, batchId):
        total = sum(m.num_rows for m in messages if m is not None)
        print(f"Committed batch {batchId} with {total} rows")

    def abort(self, messages, batchId):
        print(f"Batch {batchId} failed, performing cleanup")