DataSourceReader

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

Čtenáři zdrojů dat zodpovídají za výstup dat ze zdroje dat. Implementujte tuto třídu a vraťte instanci, DataSource.reader() aby byl zdroj dat čitelný.

Syntaxe

from pyspark.sql.datasource import DataSourceReader

class MyDataSourceReader(DataSourceReader):
    def read(self, partition):
        ...

Methods

Metoda	Description
pushFilters(filters)	Volané se seznamem filtrů, které je možné odeslat do zdroje dat. Vrátí iterable filtrů, které musí Spark vyhodnotit. Ve výchozím nastavení vrátí všechny filtry, které označují, že se nenasdílí žádné filtry. pushFilters() je povoleno upravovat self. Objekt musí po úpravě zůstat picklable. Změny, které self se mají zobrazit partitions() , a read().
partitions()	Vrátí posloupnost InputPartition objektů, které rozdělují data čtení do paralelních úloh. Ve výchozím nastavení vrátí jeden oddíl. Přepsání pro lepší výkon při čtení velkých datových sad Všechny hodnoty oddílů vrácené objekty partitions() musí být vybratelné objekty.
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.

Příklady

Implementujte základní čtečku, která vrací řádky ze seznamu oddílů:

from pyspark.sql.datasource import DataSource, DataSourceReader, InputPartition

class MyDataSourceReader(DataSourceReader):
    def partitions(self):
        return [InputPartition(1), InputPartition(2), InputPartition(3)]

    def read(self, partition):
        yield (partition.value, 0)
        yield (partition.value, 1)

Vrácení řádků pomocí PyArrow RecordBatch:

class MyDataSourceReader(DataSourceReader):
    def read(self, partition):
        import pyarrow as pa
        data = {
            "partition": [partition.value] * 2,
            "value": [0, 1]
        }
        table = pa.Table.from_pydict(data)
        for batch in table.to_batches():
            yield batch

Implementujte odsdílení filtru pro podporu EqualTo filtrů:

from pyspark.sql.datasource import DataSourceReader, EqualTo

class MyDataSourceReader(DataSourceReader):
    def __init__(self):
        self.filters = []

    def pushFilters(self, filters):
        for f in filters:
            if isinstance(f, EqualTo):
                self.filters.append(f)
            else:
                yield f

    def read(self, partition):
        ...