DataSourceStreamReader

Základní třída pro čtečky streamovaných zdrojů dat.

Čtenáři datových proudů zdroje dat zodpovídají za výstup dat ze streamovaného zdroje dat. Implementujte tuto třídu a vraťte instanci, DataSource.streamReader() aby byl zdroj dat čitelný jako zdroj streamování.

Přidáno do Databricks Runtime 15.2

Syntaxe

from pyspark.sql.datasource import DataSourceStreamReader

class MyDataSourceStreamReader(DataSourceStreamReader):
    def initialOffset(self):
        ...

    def partitions(self, start, end):
        ...

    def read(self, partition):
        ...

Methods

Metoda Description
initialOffset() Vrátí počáteční posun streamovaného zdroje dat jako dicthodnotu . Nový streamovací dotaz začne číst z tohoto posunu. Musí vrátit páry posunu klíč-hodnota primitivních typů ve formátu JSON nebo dict formátu. Vyvolá, PySparkNotImplementedError pokud není implementováno.
latestOffset(start, limit) Vrátí nejnovější dostupný posun jako dictpočáteční posun a limit pro čtení. Zdroj může vrátit stejný posun jako start v případě, že neexistují žádná nová data. Zdroj musí vždy respektovat dané limit. Musí vrátit páry posunu klíč-hodnota primitivních typů ve formátu JSON nebo dict formátu. Vyvolá, PySparkNotImplementedError pokud není implementováno.
partitions(start, end) Vrátí sekvenci InputPartition objektů představujících data mezi start a end posuny. Vrátí prázdnou sekvenci, pokud start se rovná end. Každá InputPartition představuje rozdělení dat, které je možné zpracovat jedním úkolem Sparku.
read(partition) Generuje data pro daný oddíl a vrací iterátor řazených kolekcí členů, řádků nebo objektů PyArrow RecordBatch . Každá řazená kolekce členů nebo řádek se převede na řádek v konečném datovém rámci. Tato metoda je abstraktní a musí být implementována.
commit(end) Informuje zdroj, že Spark dokončil zpracování všech dat pro posuny menší než nebo rovno end. Spark požádá pouze o posuny větší než end v budoucnu.
stop() Zastaví zdroj a uvolní všechny prostředky, které přidělil. Vyvoláno při ukončení dotazu streamování.

Poznámky

  • read() je statická a bezstavová. Nepřistupujte ke členům proměnlivé třídy nebo udržujte stav v paměti mezi různými vyvoláním read().
  • Všechny hodnoty oddílů vrácené objekty partitions() musí být vybratelné objekty.
  • Posuny jsou reprezentovány jako dict rekurzivní dict , jejichž klíče a hodnoty jsou primitivní typy: celé číslo, řetězec nebo logická hodnota.

Příklady

Implementujte čtečku streamování, která čte z posloupnosti indexovaných záznamů:

from pyspark.sql.datasource import (
    DataSource,
    DataSourceStreamReader,
    InputPartition,
)

class MyDataSourceStreamReader(DataSourceStreamReader):
    def initialOffset(self):
        return {"index": 0}

    def latestOffset(self, start, limit):
        return {"index": start["index"] + 10}

    def partitions(self, start, end):
        return [
            InputPartition(i)
            for i in range(start["index"], end["index"])
        ]

    def read(self, partition):
        yield (partition.value, f"record-{partition.value}")

    def commit(self, end):
        print(f"Committed up to offset {end}")

    def stop(self):
        print("Stopping stream reader")
  • Last updated on