PySpark özel veri kaynakları

PySpark özel veri kaynakları, Python kullanarak Apache Spark'ta özel veri kaynaklarından okuma ve özel veri havuzlarına yazma olanağı sağlayan Python (PySpark) DataSource API'sini kullanarak oluşturulur. PySpark özel veri kaynaklarını kullanarak veri sistemlerine özel bağlantılar tanımlayabilir ve yeniden kullanılabilir veri kaynakları oluşturmak için ek işlevler uygulayabilirsiniz.

Spark, Delta, Iceberg, Parquet, JSON, CSV ve JDBC gibi standart formatlar için yerleşik destek sunar, ancak REST API'leri, Google Sheets, Hugging Face veri setleri veya özel dahili hizmetler gibi birçok diğer sistem için bu destek yoktur. Python DataSource API'si bu boşluğu dolduruyor: bu sistemlere JVM tabanlı bağlantı geliştirmeden saf Python ile bağlantı kurmuyorsunuz ve bunları, Spark SQL dahil olmak üzere, herhangi bir yerleşik Spark veri kaynağı gibi kullanıyorsunuz.

Dikkat

PySpark özel veri kaynakları Databricks Runtime 15.4 LTS ve üzerini veya sunucusuz ortam sürüm 2'yi gerektirir.

DataSource sınıfı

PySpark DataSource , veri okuyucuları ve yazıcılar oluşturmak için yöntemler sağlayan bir temel sınıftır.

Veri kaynağı alt sınıfını uygulama

Kullanım örneğine bağlı olarak, veri kaynağının okunabilir, yazılabilir veya her ikisini birden yapabilmesi için aşağıdakilerin herhangi bir alt sınıf tarafından uygulanması gerekir:

Özellik veya Yöntem	Açıklama
name	Gerekli. Veri kaynağının adı
schema	Gerekli. Okunacak veya yazılacak veri kaynağının şeması
reader()	Veri kaynağının okunabilir olması için bir DataSourceReader döndürmelidir (toplu iş)
writer()	Veri havuzu yazılabilir hale getirmek için bir DataSourceWriter döndürmelidir (toplu iş)
streamReader() veya simpleStreamReader()	Veri akışını okunabilir hale getirmek için bir DataSourceStreamReader döndürmelidir (akış)
streamWriter()	Veri akışını yazılabilir hale getirmek için bir DataSourceStreamWriter döndürmelidir (akış)

Dikkat

Kullanıcı tanımlı DataSource, DataSourceReader, DataSourceWriter, DataSourceStreamReader, , DataSourceStreamWriterve yöntemleri seri hale getirilebilir olmalıdır. Başka bir deyişle, bir ilkel tür içeren bir sözlük veya iç içe geçmiş bir sözlük olmalıdır.

Veri kaynağını kaydetme

Arabirimi uyguladıktan sonra kaydetmeniz gerekir, ardından aşağıdaki örnekte gösterildiği gibi yükleyebilir veya başka bir şekilde kullanabilirsiniz:

# Register the data source
spark.dataSource.register(MyDataSourceClass)

# Read from a custom data source
spark.read.format("my_datasource_name").load().show()

Örnek 1: Toplu sorgu için PySpark DataSource oluşturma

PySpark DataSource okuyucu özelliklerini göstermek için Python paketini kullanarak faker örnek veriler oluşturan bir veri kaynağı oluşturun. hakkında fakerdaha fazla bilgi için Faker belgelerine bakın.

faker Aşağıdaki komutu kullanarak paketi yükleyin:

%pip install faker

1. Adım: Toplu sorgu için okuyucuyu uygulama

İlk olarak, örnek veriler oluşturmak için okuyucu mantığını uygulayın. Şemadaki her alanı doldurmak için yüklü faker kitaplığını kullanın.

class FakeDataSourceReader(DataSourceReader):

    def __init__(self, schema, options):
        self.schema: StructType = schema
        self.options = options

    def read(self, partition):
        # Library imports must be within the method.
        from faker import Faker
        fake = Faker()

        # Every value in this `self.options` dictionary is a string.
        num_rows = int(self.options.get("numRows", 3))
        for _ in range(num_rows):
            row = []
            for field in self.schema.fields:
                value = getattr(fake, field.name)()
                row.append(value)
            yield tuple(row)

2. Adım: DataSource örneğini tanımlama

Ardından yeni PySpark DataSource'unuzu bir ad, şema ve okuyucu ile alt sınıfı DataSource olarak tanımlayın. reader() yönteminin, bir toplu iş sorgusunda bir veri kaynağından okumak için tanımlanması gerekmektedir.

from pyspark.sql.datasource import DataSource, DataSourceReader
from pyspark.sql.types import StructType

class FakeDataSource(DataSource):
    """
    An example data source for batch query using the `faker` library.
    """

    @classmethod
    def name(cls):
        return "fake"

    def schema(self):
        return "name string, date string, zipcode string, state string"

    def reader(self, schema: StructType):
        return FakeDataSourceReader(schema, self.options)

3. Adım: Örnek veri kaynağını kaydetme ve kullanma

Veri kaynağını kullanmak için onu kaydedin. Varsayılan olarak, FakeDataSource üç satırı vardır ve şema şu string alanlarını içerir: name, date, zipcode, state. Aşağıdaki örnek, örnek veri kaynağını varsayılan değerlerle kaydeder, yükler ve çıkışını yapar:

spark.dataSource.register(FakeDataSource)
spark.read.format("fake").load().show()

+-----------------+----------+-------+----------+
|             name|      date|zipcode|     state|
+-----------------+----------+-------+----------+
|Christine Sampson|1979-04-24|  79766|  Colorado|
|       Shelby Cox|2011-08-05|  24596|   Florida|
|  Amanda Robinson|2019-01-06|  57395|Washington|
+-----------------+----------+-------+----------+

Yalnızca string alanlar desteklenir, ancak test ve geliştirme için rastgele veri oluşturmak üzere faker paket sağlayıcılarının alanlarına karşılık gelen alanlarla bir şema belirtebilirsiniz. Aşağıdaki örnek, name ve company alanlarıyla veri kaynağını yükler:

spark.read.format("fake").schema("name string, company string").load().show()

+---------------------+--------------+
|name                 |company       |
+---------------------+--------------+
|Tanner Brennan       |Adams Group   |
|Leslie Maxwell       |Santiago Group|
|Mrs. Jacqueline Brown|Maynard Inc   |
+---------------------+--------------+

Veri kaynağını özel sayıda satırla yüklemek için seçeneğini belirtin numRows . Aşağıdaki örnek 5 satır belirtir:

spark.read.format("fake").option("numRows", 5).load().show()

+--------------+----------+-------+------------+
|          name|      date|zipcode|       state|
+--------------+----------+-------+------------+
|  Pam Mitchell|1988-10-20|  23788|   Tennessee|
|Melissa Turner|1996-06-14|  30851|      Nevada|
|  Brian Ramsey|2021-08-21|  55277|  Washington|
|  Caitlin Reed|1983-06-22|  89813|Pennsylvania|
| Douglas James|2007-01-18|  46226|     Alabama|
+--------------+----------+-------+------------+

Örnek 2: Toplu sorguda özel veri yugununa yazma

PySpark DataSource yazıcı özelliklerini göstermek için, bir DataFrame’in her bölümünü bir dosyaya yazan ve ardından iş commit edildiğinde bir özet işaret dosyası yazan bir veri kaynağı oluşturun.

Adım 1: Toplu sorgu için yazarı uygulayın

Öncelikle, yazar mantığını uygulayın. Her yürütücü, her bölüm için bir kez çağrı yapar write() . Tüm yazma görevleri başarıyla tamamlandıktan sonra, sürücü commit()'yi çağırır. Görevlerden biri başarısız olursa, sürücü bunun yerine abort() çağırır.

from dataclasses import dataclass
from pyspark.sql.datasource import DataSourceWriter, WriterCommitMessage

@dataclass
class SimpleCommitMessage(WriterCommitMessage):
    partition_id: int
    count: int

class FakeDataSourceWriter(DataSourceWriter):
    def __init__(self, options):
        self.path = options.get("path")
        assert self.path is not None

    def write(self, iterator):
        """
        Writes the rows in a partition to a file, then returns a commit message with the row count. Library imports must be within the method.
        """
        import json
        import os
        from pyspark import TaskContext

        # Runs on an executor, so create the output directory on the local node.
        os.makedirs(self.path, exist_ok=True)
        partition_id = TaskContext.get().partitionId()
        count = 0
        with open(os.path.join(self.path, f"part-{partition_id}.json"), "w") as file:
            for row in iterator:
                file.write(json.dumps(row.asDict()) + "\n")
                count += 1
        return SimpleCommitMessage(partition_id=partition_id, count=count)

    def commit(self, messages):
        """
        Runs on the driver after all write tasks succeed. Writes a summary of the write to a marker file.
        """
        import json
        import os

        # Runs on the driver, so create the output directory on the local node.
        os.makedirs(self.path, exist_ok=True)
        total_rows = sum(message.count for message in messages if message is not None)
        with open(os.path.join(self.path, "_SUCCESS"), "w") as file:
            file.write(json.dumps({"partitions": len(messages), "rows": total_rows}))

    def abort(self, messages):
        """
        Runs on the driver if any write task fails. Use it to clean up partial output.
        """
        import os

        # Runs on the driver, so create the output directory on the local node.
        os.makedirs(self.path, exist_ok=True)
        with open(os.path.join(self.path, "_FAILED"), "w") as file:
            file.write("write job aborted")

Adım 2: Yazılabilir bir Veri Kaynağı Tanımlayın

Ardından, writer() uygulayan bir DataSource alt sınıfı tanımlayın. overwrite argümanı, yazma modu overwrite olduğunda True, append olduğunda ise False olur.

from pyspark.sql.datasource import DataSource
from pyspark.sql.types import StructType

class FakeSinkDataSource(DataSource):
    """
    An example writable data source that saves rows to files.
    """

    @classmethod
    def name(cls):
        return "fakesink"

    def schema(self):
        return "name string, date string, zipcode string, state string"

    def writer(self, schema: StructType, overwrite: bool):
        return FakeDataSourceWriter(self.options)

Adım 3: Kaydolun ve veri alıcısına yazın

Veri kaynağını kullanmak için onu kaydedin. Ardından, kısa adı format() öğesine ve bir çıktı dizinini path seçeneğine geçirerek buna bir DataFrame yazın. Bu örnek, Unity Catalog birimindeki bir yola yazar. <schema>, <volume> ve <catalog> öğelerini mevcut bir birimle değiştirin.

spark.dataSource.register(FakeSinkDataSource)

output_path = "/Volumes/<catalog>/<schema>/<volume>/fakesink"

df = spark.range(3).selectExpr(
    "cast(id as string) as name",
    "'2025-01-01' as date",
    "'12345' as zipcode",
    "'California' as state",
)

df.write.format("fakesink").mode("append").option("path", output_path).save()

Dikkat

Çıktı dosya sayısı, DataFrame'deki bölüm sayısına eşittir, satır sayısına değil. Bölüm sayısı, kümenin varsayılan paralelliğinden kaynaklanır; bu nedenle, bölüm sayısı satır sayısından fazlaysa bazı bölümler hiç satır almaz ve boş dosyalar oluşturur.

Örnek 3: Varyantlar kullanarak bir PySpark GitHub DataSource oluşturun

PySpark DataSource'ta varyant kullanımını göstermek için bu örnek GitHub'dan çekme isteklerini okuyan bir veri kaynağı oluşturur.

Dikkat

Değişkenler Databricks Runtime 17.1 ve üzeri sürümlerin PySpark özel veri kaynaklarıyla desteklenir.

Çeşitlemeler hakkında bilgi için bkz. Değişken verilerini sorgulama.

1. Adım: Çekme isteklerini almak için okuyucuyu uygulama

İlk olarak, belirtilen GitHub deposundan çekme isteklerini almak için okuyucu mantığını uygulayın.

class GithubVariantPullRequestReader(DataSourceReader):
    def __init__(self, options):
        self.token = options.get("token")
        self.repo = options.get("path")
        if self.repo is None:
            raise Exception(f"Must specify a repo in `.load()` method.")
        # Every value in this `self.options` dictionary is a string.
        self.num_rows = int(options.get("numRows", 10))

    def read(self, partition):
        header = {
            "Accept": "application/vnd.github+json",
        }
        if self.token is not None:
            header["Authorization"] = f"Bearer {self.token}"
        url = f"https://api.github.com/repos/{self.repo}/pulls"
        response = requests.get(url, headers=header)
        response.raise_for_status()
        prs = response.json()
        for pr in prs[:self.num_rows]:
            yield Row(
                id = pr.get("number"),
                title = pr.get("title"),
                user = VariantVal.parseJson(json.dumps(pr.get("user"))),
                created_at = pr.get("created_at"),
                updated_at = pr.get("updated_at")
            )

2. Adım: GitHub DataSource'ı tanımlama

Ardından, yeni PySpark GitHub DataSource'unuzu adı, şeması ve yöntemiyle DataSourcealt sınıfı reader() olarak tanımlayın. Şema şu alanları içerir: id, title, user, created_at, updated_at. Alan user bir değişken olarak tanımlanır.

import json
import requests

from pyspark.sql import Row
from pyspark.sql.datasource import DataSource, DataSourceReader
from pyspark.sql.types import VariantVal

class GithubVariantDataSource(DataSource):
    @classmethod
    def name(self):
        return "githubVariant"
    def schema(self):
        return "id int, title string, user variant, created_at string, updated_at string"
    def reader(self, schema):
        return GithubVariantPullRequestReader(self.options)

3. Adım: Veri kaynağını kaydetme ve kullanma

Veri kaynağını kullanmak için onu kaydedin. Aşağıdaki örnek veri kaynağını kaydeder, ardından veri kaynağını yükler ve GitHub deposu pr verilerinin üç satırını oluşturur:

spark.dataSource.register(GithubVariantDataSource)
spark.read.format("githubVariant").option("numRows", 3).load("apache/spark").display()

+---------+-----------------------------------------------------+---------------------+----------------------+----------------------+
| id      | title                                               | user                | created_at           | updated_at           |
+---------+---------------------------------------------------- +---------------------+----------------------+----------------------+
|   51293 |[SPARK-52586][SQL] Introduce AnyTimeType             |  {"avatar_url":...} | 2025-06-26T09:20:59Z | 2025-06-26T15:22:39Z |
|   51292 |[WIP][PYTHON] Arrow UDF for aggregation              |  {"avatar_url":...} | 2025-06-26T07:52:27Z | 2025-06-26T07:52:37Z |
|   51290 |[SPARK-50686][SQL] Hash to sort aggregation fallback |  {"avatar_url":...} | 2025-06-26T06:19:58Z | 2025-06-26T06:20:07Z |
+---------+-----------------------------------------------------+---------------------+----------------------+----------------------+

Örnek 4: Akış okuma ve yazma için PySpark DataSource oluşturun

PySpark DataSource akış okuyucu ve yazıcı özelliklerini göstermek için Python paketini kullanarak faker her mikrobatch'te iki satır oluşturan örnek bir veri kaynağı oluşturun. hakkında fakerdaha fazla bilgi için Faker belgelerine bakın.

faker Aşağıdaki komutu kullanarak paketi yükleyin:

%pip install faker

1. Adım: Akış okuyucuyu uygulama

İlk olarak, her mikrobatch içinde iki satır oluşturan örnek akış veri okuyucusu uygulayın. DataSourceStreamReader uygulayabilirsiniz, veya veri kaynağının aktarım hızı düşükse ve bölümleme gerekmiyorsa, bunun yerine SimpleDataSourceStreamReader uygulayabilirsiniz. Ya simpleStreamReader() ya da streamReader() uygulanmak zorundadır ve simpleStreamReader(), yalnızca streamReader() uygulanmadığında çağrılır.

DataSourceStreamReader uygulaması

streamReader örneğinin, DataSourceStreamReader arabirimiyle uygulanan her mikrobatchte 2 artan bir tamsayı uzaklığı vardır.

from pyspark.sql.datasource import InputPartition
from typing import Iterator, Tuple
import os
import json

class RangePartition(InputPartition):
    def __init__(self, start, end):
        self.start = start
        self.end = end

class FakeStreamReader(DataSourceStreamReader):
    def __init__(self, schema, options):
        self.current = 0

    def initialOffset(self) -> dict:
        """
        Returns the initial start offset of the reader.
        """
        return {"offset": 0}

    def latestOffset(self) -> dict:
        """
        Returns the current latest offset that the next microbatch will read to.
        """
        self.current += 2
        return {"offset": self.current}

    def partitions(self, start: dict, end: dict):
        """
        Plans the partitioning of the current microbatch defined by start and end offset. It
        needs to return a sequence of :class:`InputPartition` objects.
        """
        return [RangePartition(start["offset"], end["offset"])]

    def commit(self, end: dict):
        """
        This is invoked when the query has finished processing data before end offset. This
        can be used to clean up the resource.
        """
        pass

    def read(self, partition) -> Iterator[Tuple]:
        """
        Takes a partition as an input and reads an iterator of tuples from the data source.
        """
        start, end = partition.start, partition.end
        for i in range(start, end):
            yield (i, str(i))

SimpleDataSourceStreamReader uygulaması

SimpleStreamReader örneği, her toplu işlemde iki satır oluşturan ancak FakeStreamReader arabirimi ile bölümleme olmadan uygulanan SimpleDataSourceStreamReader örneğiyle aynıdır.

class SimpleStreamReader(SimpleDataSourceStreamReader):
    def initialOffset(self):
        """
        Returns the initial start offset of the reader.
        """
        return {"offset": 0}

    def read(self, start: dict) -> (Iterator[Tuple], dict):
        """
        Takes start offset as an input, then returns an iterator of tuples and the start offset of the next read.
        """
        start_idx = start["offset"]
        it = iter([(i,) for i in range(start_idx, start_idx + 2)])
        return (it, {"offset": start_idx + 2})

    def readBetweenOffsets(self, start: dict, end: dict) -> Iterator[Tuple]:
        """
        Takes start and end offset as inputs, then reads an iterator of data deterministically.
        This is called when the query replays batches during restart or after a failure.
        """
        start_idx = start["offset"]
        end_idx = end["offset"]
        return iter([(i,) for i in range(start_idx, end_idx)])

    def commit(self, end):
        """
        This is invoked when the query has finished processing data before end offset. This can be used to clean up resources.
        """
        pass

2. Adım: Akış yazıcısını uygulama

Ardından streaming writer'ı uygulayın. Bu akış veri yazıcısı, her mikro veri kümesinin meta verilerini yerel bir yola yazar.

from pyspark.sql.datasource import DataSourceStreamWriter, WriterCommitMessage

class SimpleCommitMessage(WriterCommitMessage):
   def __init__(self, partition_id: int, count: int):
       self.partition_id = partition_id
       self.count = count

class FakeStreamWriter(DataSourceStreamWriter):
   def __init__(self, options):
       self.options = options
       self.path = self.options.get("path")
       assert self.path is not None

   def write(self, iterator):
       """
       Writes the data and then returns the commit message for that partition. Library imports must be within the method.
       """
       from pyspark import TaskContext
       context = TaskContext.get()
       partition_id = context.partitionId()
       cnt = 0
       for row in iterator:
           cnt += 1
       return SimpleCommitMessage(partition_id=partition_id, count=cnt)

   def commit(self, messages, batchId) -> None:
       """
       Receives a sequence of :class:`WriterCommitMessage` when all write tasks have succeeded, then decides what to do with it.
       In this FakeStreamWriter, the metadata of the microbatch(number of rows and partitions) is written into a JSON file inside commit().
       """
       status = dict(num_partitions=len(messages), rows=sum(m.count for m in messages))
       with open(os.path.join(self.path, f"{batchId}.json"), "a") as file:
           file.write(json.dumps(status) + "\n")

   def abort(self, messages, batchId) -> None:
       """
       Receives a sequence of :class:`WriterCommitMessage` from successful tasks when some other tasks have failed, then decides what to do with it.
       In this FakeStreamWriter, a failure message is written into a text file inside abort().
       """
       with open(os.path.join(self.path, f"{batchId}.txt"), "w") as file:
           file.write(f"failed in batch {batchId}")

3. Adım: DataSource örneğini tanımlama

Şimdi yeni PySpark DataSource'unuzu bir ad, şema, yöntemler DataSource ve streamReader()ile alt sınıfı streamWriter() olarak tanımlayın.

from pyspark.sql.datasource import DataSource, DataSourceStreamReader, SimpleDataSourceStreamReader, DataSourceStreamWriter
from pyspark.sql.types import StructType

class FakeStreamDataSource(DataSource):
    """
    An example data source for streaming read and write using the `faker` library.
    """

    @classmethod
    def name(cls):
        return "fakestream"

    def schema(self):
        return "name string, state string"

    def streamReader(self, schema: StructType):
        return FakeStreamReader(schema, self.options)

    # If you don't need partitioning, you can implement the simpleStreamReader method instead of streamReader.
    # def simpleStreamReader(self, schema: StructType):
    #    return SimpleStreamReader()

    def streamWriter(self, schema: StructType, overwrite: bool):
        return FakeStreamWriter(self.options)

4. Adım: Örnek veri kaynağını kaydetme ve kullanma

Veri kaynağını kullanmak için onu kaydedin. Kaydedildikten sonra akış sorgularında kaynak veya havuz olarak kullanmak için format()öğesine kısa bir ad veya tam ad geçirebilirsiniz. Aşağıdaki örnek, veri kaynağını kaydeder, ardından örnek veri kaynağından okuyan ve konsola çıkış veren bir sorgu başlatır:

spark.dataSource.register(FakeStreamDataSource)
query = spark.readStream.format("fakestream").load().writeStream.format("console").start()

Alternatif olarak, aşağıdaki kod örnek akışı havuz olarak kullanır ve bir çıkış yolu belirtir:

spark.dataSource.register(FakeStreamDataSource)

# Make sure the output directory exists and is writable
output_path = "/output_path"
dbutils.fs.mkdirs(output_path)
checkpoint_path = "/output_path/checkpoint"

query = (
    spark.readStream
    .format("fakestream")
    .load()
    .writeStream
    .format("fakestream")
    .option("path", output_path)
    .option("checkpointLocation", checkpoint_path)
    .start()
)

Örnek 5: Google BigQuery yayın bağlantısı oluşturun

Aşağıdaki örnekte, PySpark DataSource kullanarak Google BigQuery (BQ) için özel akış bağlayıcısının nasıl derlenmesi gösterilmektedir. Databricks, BigQuery toplu alımı için bir Spark bağlayıcısı sağlar ve Lakehouse Federation ayrıca herhangi bir BigQuery veri kümesine uzaktan bağlanabilir ve yabancı katalog oluşturma yoluyla veri çekebilir, ancak her ikisi de artımlı veya sürekli akış iş akışlarını tam olarak desteklemez. Bu bağlayıcı, kalıcı denetim noktası kullanılarak akış kaynakları tarafından beslenen BigQuery tablolarından aşamalar halinde artımlı veri geçişi ve neredeyse gerçek zamanlı veri geçişi sağlar.

Bu özel bağlayıcı aşağıdaki özelliklere sahiptir:

• Yapılandırılmış Akış ve Lakeflow işlem hatları ile uyumludur.
• Artımlı kayıt izlemeyi ve sürekli akış alımını destekler ve Yapılandırılmış Akış semantiğini izler.
• Daha hızlı, daha ucuz veri iletimi için RPC tabanlı bir protokolle BigQuery Depolama API'sini kullanır.
• Geçirilen tabloları doğrudan Unity Kataloğu'na yazar.
• Tarih veya zaman damgası tabanlı artımlı bir alan kullanarak denetim noktalarını otomatik olarak yönetir.
• ile Trigger.AvailableNow()toplu alımı destekler.
• Ara bulut depolama alanı gerektirmez.
• Ok veya Avro biçimini kullanarak BigQuery veri iletimini serileştirir.
• Otomatik paralellik yönetimi yapar ve veri hacmine göre işi Spark çalışanlarına dağıtır.
• ScD Tür 1 veya Tür 2 desenlerini kullanan Gümüş ve Altın katmanı geçişleri desteğiyle BigQuery'den Ham ve Bronz katman geçişi için uygundur.

Önkoşullar

Özel bağlayıcıyı uygulamadan önce gerekli paketleri yükleyin:

%pip install faker google.cloud google.cloud.bigquery google.cloud.bigquery_storage

1. Adım: Akış okuyucuyu uygulama

İlk olarak akış veri okuyucusu uygulayın. Alt sınıfın DataSourceStreamReader aşağıdaki yöntemleri uygulaması gerekir:

• initialOffset(self) -> dict
• latestOffset(self) -> dict
• partitions(self, start: dict, end: dict) -> Sequence[InputPartition]
• read(self, partition: InputPartition) -> Union[Iterator[Tuple], Iterator[Row]]
• commit(self, end: dict) -> None
• stop(self) -> None

Her yöntemle ilgili ayrıntılar için bkz. Yöntemler.

import os
from pyspark.sql.datasource import DataSourceStreamReader, InputPartition
from pyspark.sql.datasource import DataSourceStreamWriter
from pyspark.sql import Row
from pyspark.sql import SparkSession
from pyspark.sql.datasource import DataSource
from pathlib import Path
from pyarrow.lib import TimestampScalar
from datetime import datetime
from typing import Iterator, Tuple, Any, Dict, List, Sequence
from google.cloud.bigquery_storage import BigQueryReadClient, ReadSession
from google.cloud import bigquery_storage
import pandas
import datetime
import uuid
import time, logging

start_time = time.time()


class RangePartition(InputPartition):
    def __init__(self, session: ReadSession, stream_idx: int):
        self.session = session
        self.stream_idx = stream_idx


class BQStreamReader(DataSourceStreamReader):

    def __init__(self, schema, options):
        self.project_id = options.get("project_id")
        self.dataset = options.get("dataset")
        self.table = options.get("table")
        self.json_auth_file = "/home/"+options.get("service_auth_json_file_name")
        self.max_parallel_conn = options.get("max_parallel_conn", 1000)
        self.incremental_checkpoint_field = options.get("incremental_checkpoint_field", "")

        self.last_offset = None

    def initialOffset(self) -> dict:
        """
        Returns the initial start offset of the reader.
        """
        from datetime import datetime
        logging.info("Inside initialOffset!!!!!")
        # self.increment_latest_vals.append(datetime.strptime('1900-01-01 23:57:12', "%Y-%m-%d %H:%M:%S"))
        self.last_offset = '1900-01-01 23:57:12'

        return {"offset": str(self.last_offset)}

    def latestOffset(self):
        """
        Returns the current latest offset that the next microbatch will read to.
        """
        from datetime import datetime
        from google.cloud import bigquery

        if (self.last_offset is None):
            self.last_offset = '1900-01-01 23:57:12'

        client = bigquery.Client.from_service_account_json(self.json_auth_file)
        # max_offset=start["offset"]
        logging.info(f"************************last_offset: {self.last_offset}***********************")
        f_sql_str = ''
        for x_str in self.incremental_checkpoint_field.strip().split(","):
            f_sql_str += f"{x_str}>'{self.last_offset}' or "
        f_sql_str = f_sql_str[:-3]
        job_query = client.query(
            f"select max({self.incremental_checkpoint_field}) from {self.project_id}.{self.dataset}.{self.table} where {f_sql_str}")
        for query in job_query.result():
            max_res = query[0]

        if (str(max_res).lower() != 'none'):
            return {"offset": str(max_res)}

        return {"offset": str(self.last_offset)}

    def partitions(self, start: dict, end: dict) -> Sequence[InputPartition]:

        """
        Plans the partitioning of the current microbatch defined by start and end offset. It
        needs to return a sequence of :class:`InputPartition` objects.
        """
        if (self.last_offset is None):
            self.last_offset = end['offset']

        os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = self.json_auth_file

        # project_id = self.auth_project_id

        client = BigQueryReadClient()

        # This example reads baby name data from the public datasets.
        table = "projects/{}/datasets/{}/tables/{}".format(
            self.project_id, self.dataset, self.table
        )
        requested_session = bigquery_storage.ReadSession()
        requested_session.table = table
        if (self.incremental_checkpoint_field != ''):
            start_offset = start["offset"]
            end_offset = end["offset"]
            f_sql_str = ''
            for x_str in self.incremental_checkpoint_field.strip().split(","):
                f_sql_str += f"({x_str}>'{start_offset}' and {x_str}<='{end_offset}') or "
            f_sql_str = f_sql_str[:-3]
            requested_session.read_options.row_restriction = f"{f_sql_str}"

        # This example leverages Apache Avro.
        requested_session.data_format = bigquery_storage.DataFormat.AVRO

        parent = "projects/{}".format(self.project_id)
        session = client.create_read_session(
            request={
                "parent": parent,
                "read_session": requested_session,
                "max_stream_count": int(self.max_parallel_conn),
            },
        )
        self.last_offset = end['offset']
        return [RangePartition(session, i) for i in range(len(session.streams))]

    def read(self, partition) -> Iterator[List]:
        """
        Takes a partition as an input and reads an iterator of tuples from the data source.
        """
        from datetime import datetime
        session = partition.session
        stream_idx = partition.stream_idx
        os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = self.json_auth_file
        client_1 = BigQueryReadClient()
        # requested_session.read_options.selected_fields = ["census_tract", "clearance_date", "clearance_status"]
        reader = client_1.read_rows(session.streams[stream_idx].name)
        reader_iter = []

        for message in reader.rows():
            reader_iter_in = []
            for k, v in message.items():
                reader_iter_in.append(v)
            # yield(reader_iter)
            reader_iter.append(reader_iter_in)
            # yield (message['hash'], message['size'], message['virtual_size'], message['version'])
        # self.increment_latest_vals.append(max_incr_val)
        return iter(reader_iter)

    def commit(self, end):

        """
        This is invoked when the query has finished processing data before end offset. This
        can be used to clean up the resource.
        """
        pass

2. Adım: DataSource'ı tanımlama

Ardından özel veri kaynağını tanımlayın. Alt sınıfın DataSource aşağıdaki yöntemleri uygulaması gerekir:

• name(cls) -> str
• schema(self) -> Union[StructType, str]

Her yöntemle ilgili ayrıntılar için bkz. Yöntemler.

from pyspark.sql.datasource import DataSource
from pyspark.sql.types import StructType
from google.cloud import bigquery

class BQStreamDataSource(DataSource):
    """
    An example data source for streaming data from a public API containing users' comments.
    """

    @classmethod
    def name(cls):
        return "bigquery-streaming"

    def schema(self):
        type_map = {'integer': 'long', 'float': 'double', 'record': 'string'}
        json_auth_file = "/home/" + self.options.get("service_auth_json_file_name")
        client = bigquery.Client.from_service_account_json(json_auth_file)
        table_ref = self.options.get("project_id") + '.' + self.options.get("dataset") + '.' + self.options.get("table")
        table = client.get_table(table_ref)
        original_schema = table.schema
        result = []
        for schema in original_schema:
            col_attr_name = schema.name
            if (schema.mode != 'REPEATED'):
                col_attr_type = type_map.get(schema.field_type.lower(), schema.field_type.lower())
            else:
                col_attr_type = f"array<{type_map.get(schema.field_type.lower(), schema.field_type.lower())}>"
            result.append(col_attr_name + " " + col_attr_type)

        return ",".join(result)
        # return "census_tract double,clearance_date string,clearance_status string"

    def streamReader(self, schema: StructType):
        return BQStreamReader(schema, self.options)

3. Adım: Akış sorgusunu yapılandırma ve başlatma

Son olarak bağlayıcıyı kaydedin, ardından akış sorgusunu yapılandırın ve başlatın:

spark.dataSource.register(BQStreamDataSource)

# Ingests table data incrementally using the provided timestamp-based field.
# The latest value is checkpointed using offset semantics.
# Without the incremental input field, full table ingestion is performed.
# Service account JSON files must be available to every Spark executor worker
# in the /home folder using --files /home/<file_name>.json or an init script.

query = (
    spark.readStream.format("bigquery-streaming")
    .option("project_id", <bq_project_id>)
    .option("incremental_checkpoint_field", <table_incremental_ts_based_col>)
    .option("dataset", <bq_dataset_name>)
    .option("table", <bq_table_name>)
    .option("service_auth_json_file_name", <service_account_json_file_name>)
    .option("max_parallel_conn", <max_parallel_threads_to_pull_data>)  # defaults to max 1000
    .load()
)

(
    query.writeStream.trigger(processingTime="30 seconds")
    .option("checkpointLocation", "checkpoint_path")
    .foreachBatch(writeToTable)  # your target table write function
    .start()
)

Yürütme sırası

Özel akımın fonksiyon yürütme sırası aşağıda açıklanmıştır.

Spark Stream DataFrame'i yüklemek için:

name(cls)
schema()

Yeni bir sorgunun mikrobatch (n) başlangıcında veya mevcut bir sorguyu yeniden başlatırken (yeni veya mevcut kontrol noktası):

partitions(end_offset, end_offset)  # loads the last saved offset from the checkpoint at query restart
latestOffset()
partitions(start_offset, end_offset)  # plans partitions and distributes to Python workers
read()  # user’s source read definition, runs on each Python worker
commit()

Var olan bir denetim noktasında çalışan bir sorgunun sonraki (n+1) mikrobatch'i için:

latestOffset()
partitions(start_offset, end_offset)
read()
commit()

Dikkat

İşlev, latestOffset denetim noktası oluşturmayı düzenler. İlkel türde bir denetim noktası değişkenini işlevler arasında paylaşın ve sözlük olarak döndürebilirsiniz. Örneğin: return {"offset": str(self.last_offset)}

Örnek 6: Hardış bir API ile kimlik doğrulama

Bu örnekte, veri kaynağı kodunun hiçbir zaman sabit kodlanmış belirteçler veya kimlik bilgileri içermemesi için Unity Kataloğu HTTP bağlantısı kullanılarak dış HTTP API'siyle PySpark veri kaynağının kimliğini doğrulama işlemi gösterilmektedir.

Dikkat

Unity Kataloğu HTTP bağlantısı için kimlik bilgisi ekleme özelliği, Databricks Runtime 18.1 veya üzerini gerektirir.

1. Adım: HTTP bağlantısı oluşturma

Veri kaynağını uygulamadan önce Unity Kataloğu'nda adlı my_weather_api bir HTTP bağlantısı oluşturun ve kullanıcılara veya gruplara MANAGE bu bağlantı için izin verin. Yalnızca bağlantı üzerinde izni olan MANAGE kullanıcılar kimlik bilgisi ekleme işlemini tetikleyebilir.

API belirtecini bir Databricks gizlisi olarak depolayın ve belirtecin düz metin değerini girmek yerine buna secret işleviyle başvurun; böylece kimlik bilgileri bağlantı tanımında hiçbir zaman görünmez.

CREATE CONNECTION my_weather_api TYPE HTTP
OPTIONS (
    host 'https://api.openweathermap.org',
    base_path '/data/2.5',
    bearer_token secret('my_secret_scope', 'weather_api_token')
);

GRANT MANAGE ON CONNECTION my_weather_api TO `user@example.com`;

2. Adım: Toplu sorgu için okuyucuyu uygulama

Ardından, REST API'den veri getirmek için okuyucu mantığını uygulayın. Okuyucu, eklenen host, base_pathve bearer_token değerlerini seçeneklerinden okur, böylece kodda hiçbir kimlik bilgisi görünmez.

from pyspark.sql.datasource import DataSource, DataSourceReader, InputPartition
from urllib.parse import quote
import urllib.error
import urllib.request
import json

class WeatherApiReader(DataSourceReader):
    def __init__(self, options):
        self.host = options["host"]
        self.base_path = options["base_path"]
        self.token = options["bearer_token"]
        # Every value in this `options` dictionary is a string.
        self.cities = options.get("cities", "Seattle,Portland,Denver").split(",")

    def partitions(self):
        return [InputPartition(city.strip()) for city in self.cities]

    def read(self, partition):
        city = partition.value
        # URL-encode the city so names with spaces or non-ASCII characters (for example, "New York" or "São Paulo") produce a valid query string.
        url = f"{self.host}{self.base_path}/weather?q={quote(city)}&units=metric"
        req = urllib.request.Request(url)
        req.add_header("Authorization", f"Bearer {self.token}")
        try:
            # Set a timeout so a slow or unresponsive API surfaces a controlled error instead of hanging the Spark task.
            with urllib.request.urlopen(req, timeout=30) as resp:
                data = json.loads(resp.read().decode())
        except (urllib.error.URLError, TimeoutError) as e:
            raise RuntimeError(f"Weather API request failed for {city}: {e}")
        # Validate the response shape before indexing so an error payload raises a clear message instead of a KeyError.
        try:
            main = data["main"]
            weather = data["weather"][0]
        except (KeyError, IndexError, TypeError):
            raise RuntimeError(f"Unexpected weather API response for {city}: {data}")
        yield (city, main["temp"], main["humidity"], weather["description"])

3. Adım: DataSource örneğini tanımlama

Şimdi yeni PySpark DataSource'unuzu bir ad, şema ve okuyucu ile alt sınıfı DataSource olarak tanımlayın.

class WeatherApiSource(DataSource):
    def __init__(self, options):
        self.options = options

    @classmethod
    def name(cls):
        return "weather_api"

    def schema(self):
        return "city STRING, temperature DOUBLE, humidity INT, description STRING"

    def reader(self, schema):
        return WeatherApiReader(self.options)

4. Adım: Veri kaynağını kaydetme ve kullanma

Veri kaynağını kullanmak için onu kaydedin. Ardından Unity Catalog HTTP bağlantısına databricks.connection seçeneğiyle başvurun. Spark sürücüsü, Unity Kataloğu'ndan kısa ömürlü OAuth2 kimlik bilgilerini otomatik olarak alır ve bunları (örneğin, bearer_token, hostve base_path) veri kaynağı seçenekleri haritasına ekler. Unity Kataloğu tarafından eklenen kimlik bilgileri anahtarları geçersiz kılınamaz ve ve hostgibi port genel olarak engellenen seçenekler engellenmiş olarak kalır ve kullanıcılar tarafından ayarlanamaz.

spark.dataSource.register(WeatherApiSource)

df = (
    spark.read.format("weather_api")
    .option("databricks.connection", "my_weather_api")   # Unity Catalog injects host, base_path, bearer_token
    .option("cities", "Seattle,Portland,Denver")         # user-defined option passes through
    .load()
)
df.show()

Bu örnek yalnızca toplu okuma işlemlerini uygular. Aynı databricks.connection seçeneği, veri kaynağınız ilgili yöntemleri uyguladığında akış okuma ve yazma işlemleri için de geçerlidir (akış okumaları için streamReader veya simpleStreamReader, yazma işlemleri içinse writer veya streamWriter).

Ek kaynaklar

Apache Spark topluluğu, kendi veri kaynaklarınızı oluştururken referans olarak kullanabileceğiniz örnek bağlayıcılar bulunduruyor. Bu depolar topluluk tarafından korunur ve Databricks tarafından desteklenmez:

• pyspark-data-sources: PySpark özel veri kaynağı bağlayıcılarının örneklerinden oluşan bir koleksiyon.
• pyspark_huggingface: Hugging Face veri setlerini okumak için özel bir veri kaynağı konnektörü.

Sorun giderme

Çıkış aşağıdaki hataysa, işleminiz PySpark özel veri kaynaklarını desteklemez. Databricks Runtime 15.2 veya üzerini kullanmanız gerekir.

Error: [UNSUPPORTED_FEATURE.PYTHON_DATA_SOURCE] The feature is not supported: Python data sources. SQLSTATE: 0A000