SimpleDataSourceStreamReader

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

DataSourceStreamReaderOproti tomu SimpleDataSourceStreamReader nevyžaduje plánování datových oddílů. Metoda read() umožňuje čtení dat a plánování nejnovějšího posunu ve stejnou dobu.

Vzhledem k tomu SimpleDataSourceStreamReader , že čte záznamy v ovladači Spark k určení koncového posunu každé dávky bez dělení, je vhodné pouze pro případy zjednodušeného použití, kdy je vstupní rychlost a velikost dávky malá. Používá se DataSourceStreamReader , když je propustnost čtení vysoká a není možné ji zpracovat jedním procesem.

Přidáno do Databricks Runtime 15.3

Syntaxe

from pyspark.sql.datasource import SimpleDataSourceStreamReader

class MyStreamReader(SimpleDataSourceStreamReader):
    def initialOffset(self):
        return {"offset": 0}

    def read(self, start):
        ...

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

Methods

Metoda	Description
initialOffset()	Vrátí počáteční posun streamovaného zdroje dat. Nový streamovací dotaz začne číst z tohoto posunu.
read(start)	Přečte všechna dostupná data od počátečního posunu a vrátí řazenou kolekci členů iterátoru záznamů a koncový posun pro další pokus o čtení.
readBetweenOffsets(start, end)	Čte všechna dostupná data mezi určitým počátečním a koncovým posunem. Vyvoláno během zotavení po selhání pro opakované čtení dávky deterministicky.
commit(end)	Informuje zdroj, že Spark dokončil zpracování všech dat pro posuny menší než nebo rovno end.

Příklady

Definujte vlastní zjednodušenou čtečku streamovaných zdrojů dat:

from pyspark.sql.datasource import DataSource, SimpleDataSourceStreamReader

class MyStreamingDataSource(DataSource):
    @classmethod
    def name(cls):
        return "my_streaming_source"

    def schema(self):
        return "value STRING"

    def simpleStreamReader(self, schema):
        return MySimpleStreamReader()

class MySimpleStreamReader(SimpleDataSourceStreamReader):
    def initialOffset(self):
        return {"partition-1": {"index": 0}}

    def read(self, start):
        end = {"partition-1": {"index": start["partition-1"]["index"] + 1}}
        def records():
            yield ("hello",)
        return records(), end

    def readBetweenOffsets(self, start, end):
        def records():
            yield ("hello",)
        return records()

    def commit(self, end):
        pass

spark.dataSource.register(MyStreamingDataSource)
df = spark.readStream.format("my_streaming_source").load()