DataSourceReader

Az adatforrás-olvasók alaposztálya.

Az adatforrás-olvasók felelősek az adatforrásból származó adatok kimenetéért. Implementálja ezt az osztályt, és adjon vissza egy példányt DataSource.reader() az adatforrás olvashatóvá tétele érdekében.

Szemantika

from pyspark.sql.datasource import DataSourceReader

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

Metódusok

Módszer Leírás
pushFilters(filters) Az adatforrásba leküldhető szűrők listájával hívható meg. Olyan szűrőket ad vissza, amelyeket a Sparknak még ki kell értékelnie. Alapértelmezés szerint az összes szűrőt visszaadja, ami azt jelzi, hogy a rendszer nem küld le szűrőket. pushFilters() lehetőség van a módosításra self. Az objektumnak a módosítás után választhatónak kell maradnia. A módosítások a következőn self láthatók: partitions() és read().
partitions() Olyan objektumsorozatot InputPartition ad vissza, amely párhuzamos feladatokra osztja fel az adatolvasást. Alapértelmezés szerint egyetlen partíciót ad vissza. Nagyobb adathalmazok olvasása esetén felülbírálás a jobb teljesítmény érdekében. A visszaadott partitions() partícióértékeknek választható objektumoknak kell lenniük.
read(partition) Adatokat hoz létre egy adott partícióhoz, és visszaadja a csuplok, sorok vagy PyArrow-objektumok RecordBatch iterátorát. A rendszer minden egyes rekordot vagy sort a végső DataFrame egy sorává alakít át. Ez a módszer absztrakt, és implementálandó.

Examples

Implementáljon egy alapszintű olvasót, amely a partíciók listájából ad vissza sorokat:

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)

Sorok visszaadása a PyArrow RecordBatchhasználatával:

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

Szűrőleküldés implementálása a szűrők támogatásához EqualTo :

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):
        ...
  • Last updated on