DataSourceWriter

Základní třída pro zapisovače zdrojů dat.

Za ukládání dat do zdroje dat zodpovídají za zápisy dat. Implementujte tuto třídu a vraťte instanci, DataSource.writer() aby byl zdroj dat zapisovatelný.

Přidání do Databricks Runtime 14.3 LTS

Syntaxe

from pyspark.sql.datasource import DataSourceWriter

class MyDataSourceWriter(DataSourceWriter):
    def write(self, iterator):
        ...

Methods

Metoda	Description
write(iterator)	Zapisuje data do zdroje dat. Volá se jednou na každém exekutoru. Přijímá iterátor Row objektů a vrací WriterCommitMessagezprávu , nebo None pokud neexistuje žádná zpráva potvrzení. Tato metoda je abstraktní a musí být implementována.
commit(messages)	Potvrdí úlohu zápisu pomocí seznamu zpráv potvrzení shromážděných ze všech exekutorů. Vyvoláno na ovladači, když se všechny úlohy úspěšně spustí.
abort(messages)	Přeruší úlohu zápisu 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.

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().

Příklady

Implementujte základní zapisovač, který ukládá řádky do souboru:

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

@dataclass
class MyCommitMessage(WriterCommitMessage):
    num_rows: int

class MyDataSourceWriter(DataSourceWriter):
    def __init__(self, options):
        self.path = options.get("path")

    def write(self, iterator):
        rows = list(iterator)
        with open(self.path, "w") as f:
            for row in rows:
                f.write(str(row) + "\n")
        return MyCommitMessage(num_rows=len(rows))

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

    def abort(self, messages):
        print("Write job failed, performing cleanup")