DataSourceReader

Veri kaynağı okuyucuları için temel sınıf.

Veri kaynağı okuyucuları bir veri kaynağından veri çıkışından sorumludur. Bu sınıfı uygulayın ve bir veri kaynağını okunabilir hale getirmek için öğesinden DataSource.reader() bir örnek döndürür.

Sözdizimi

from pyspark.sql.datasource import DataSourceReader

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

Methods

Yöntem Açıklama
pushFilters(filters) Veri kaynağına gönderilebilen filtrelerin listesiyle birlikte çağrılır. Spark tarafından yine değerlendirilmesi gereken filtreleri yinelenebilir bir şekilde döndürür. Varsayılan olarak, hiçbir filtrenin aşağı gönderilmediğini gösteren tüm filtreleri döndürür. pushFilters() öğesini değiştirmesine selfizin verilir. Nesne, değişiklik sonrasında picklable olarak kalmalıdır. 'de self yapılan değişiklikler ve partitions()için read() görünür.
partitions() Veri okumayı InputPartition paralel görevlere bölen bir nesne dizisi döndürür. Varsayılan olarak, tek bir bölüm döndürür. Büyük veri kümelerini okurken daha iyi performans için geçersiz kılın. tarafından partitions() döndürülen tüm bölüm değerleri seçilebilir nesneler olmalıdır.
read(partition) Belirli bir bölüm için veri oluşturur ve tanımlama kümeleri, satırlar veya PyArrow RecordBatch nesnelerinin yineleyicisini döndürür. Her tanımlama grubu veya satır, son DataFrame'deki bir satıra dönüştürülür. Bu yöntem soyut ve uygulanması gerekir.

Örnekler

Bölüm listesinden satır döndüren temel bir okuyucu uygulayın:

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)

PyArrow RecordBatchkullanarak satırları döndürme:

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

Filtreleri desteklemek EqualTo için filtre gönderimi uygulayın:

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