İşlem hatlarına veri yükleme

İşlem hatlarını kullanarak Azure Databricks üzerinde Apache Spark tarafından desteklenen herhangi bir veri kaynağından veri yükleyebilirsiniz. Bir işlem hattında, akış DataFrame’leri ve Pandas for Spark DataFrame’leri dahil olmak üzere, Spark DataFrame döndüren herhangi bir sorgu için veri kümeleri (tablolar ve görünümler) tanımlayabilirsiniz. Databricks, veri alımı görevleri için çoğu kullanım örneğinde akış tablolarının kullanılmasını önerir. Akış tabloları, Auto Loader kullanarak bulut nesne depolama birimlerinden ya da Kafka gibi ileti veri yollarından veri almak için kullanışlıdır. Veri alımı için birincil veri kümesi türü olan akış tabloları hakkında daha fazla bilgi için bkz . Akış tabloları.

Tüm veri kaynaklarının aktarım için SQL desteği bulunmamaktadır. Ancak gerektiğinde Python kullanmak için SQL ve Python kaynaklarını aynı işlem hattında karıştırabilirsiniz. Varsayılan olarak işlem hatları ile paketlenmiş olmayan kitaplıklarla çalışma hakkında ayrıntılı bilgi için bkz. İşlem hatları için Python bağımlılıklarını yönetme. Azure Databricks'te veri girişi hakkında genel bilgi için daha fazla bilgi edinmek üzere Lakeflow Connect'te standart bağlayıcılar bölümüne bakın.

Aşağıdaki örneklerde bazı yaygın veri yükleme desenleri gösterilmektedir.

Veri kaynaklarınızı ve bağlantı yolunuzu belirleyin

Boru hattı kodu yazmadan önce, verilerin geldiği her yeri envantere edin. Her kaynak için, verilerini nasıl sunduğunu (dosyalar, bir veritabanı, hizmet olarak yazılım (SaaS) sistemi, bir API veya bir akış aracılığıyla), ne sıklıkta değiştiğini ve hangi kimlik bilgilerine ve ağ erişimine ihtiyaç duyduğunu not edin. Bağlantı yöntemi genellikle bir kaynağın doğal olarak toplu mi yoksa akış mı olduğunu belirler, bu yüzden bunu erken yapmak daha sonra yeniden çalışmayı önlemeye yardımcı olur.

Her bir kaynağı aşağıdaki bağlantı yollarından birinde sınıflandırın. Aşağıdaki tablo, her boru hattı kaynağı için tercih edilen mekanizmayı listeler:

Source Bağlantı yolu
Bulut nesne depolamaya (S3, Azure Data Lake Storage, GCS) ulaşan dosyalar En yaygın başlangıç noktası. Otomatik Yükleyici (cloudFiles format) kullanın; bu cihaz artımlı keşif, şema çıkarımı ve şema evrimini yönetir. Bkz . Dosyaları bulut nesne depolama alanından yükleme.
Veritabanları ve SaaS uygulamaları (Salesforce, SQL Server, PostgreSQL, Workday) Kaynağınız için mevcutsa Lakeflow Connect yönetilen bir bağlayıcı kullanın. Yönetilen bağlayıcılar yapılandırmaya dayalıdır ve kimlik doğrulamayı ile artımlı veya CDC veri çekmeyi sizin adınıza gerçekleştirir. Bkz. Lakeflow Connect'te yönetilen bağlayıcılar. Kaynağınız için yönetilen bir bağlayıcı yoksa, verileri doğrudan içeri aktarın veya yanıtlarını önce dosya olarak kaydedin. Bkz. Pipelines'ta API'den veri alımı.
Mesaj veri yolları (Kafka, Kinesis, Azure Event Hubs, Pub/Sub) Doğrudan Yapılandırılmış Akış kaynağı olarak okunabilir çünkü bunlar yerel yayın kaynaklarıdır. Bkz. Mesaj veri yolundan veri yükle.
Diğer Delta tabloları veya Unity Catalog varlıkları, diğer işlem hatları veya görevler tarafından oluşturulan tablolar da dahil olmak üzere Doğrudan referans verin ve Unity Catalog yönetimi ile soyunun keşif ve erişimi halletmesine izin verin. Mevcut bir tablodan yükleme bölümünü görebilirsiniz.
Küçük veya statik referans verileri (arama dosyaları, nadiren değişen CSV'ler) Maddeleştirilmiş bir görünümde toplu kaynak olarak yükleyin. Neredeyse hiç değişmeyen bir şeyi yayınlamanın hiçbir faydası yok. Bkz . Bulut nesne depolamasından küçük veya statik veri setlerini yükle.
Yönetilen bir bağlayıcı olmayan rastgele bir HTTP veya REST API Verileri ardışık düzende API'den çekin veya yanıtlarını önce dosya olarak kaydedin. Bkz. Pipelines'ta API'den veri alımı.

Her kaynak için, inşa etmeden önce aşağıdakileri onaylayın:

  • Kimlik: İşlem hattının hangi kimlikle çalıştığı. Pipeline'lar bir servis ana kuruluşu olarak çalışabilir, bu yüzden önce bunu ayarlayın, böylece kişisel hesaba bağımlı olmamak için.
  • Ağ yolu: Kaynağın ihtiyaç duyduğu bağlantı; örneğin bir depolama kimlik bilgisi, harici bir konum veya Lakeflow Connect tarafından yönetilen bağlantı.
  • Değişiklik semantiği: Kaynağın güncellemeleri ve silmeleri nasıl bildirdiği, eğer bildiriyorsa. Bu, CDC'ye ihtiyacınız olup olmadığını veya kaynağı sadece ek olarak mı ele alacağınızı belirler.

Bir dosya formatı ve depolama katmanı seçin

Boru hatları bu kararın çoğunu sizin için verir. Bir boru hattının oluşturduğu her akış tablosu ve maddeleştirilmiş görünüm varsayılan olarak bir Delta tablosu olarak saklanır; bu da size ACID işlemleri, şema uygulaması ve evrimi, zaman yolculuğu ve her veri setinde Unity Kataloğu yönetişimi ve soy hakkı sağlar. Boru hattı çıktıları için formatı seçmezsiniz. Asıl kararlarınız iş hattının iki kenarında alınır:

  • Ham giriş formatı: Kaynağın ürettiği her şey, örneğin CSV, JSON veya Parquet. Auto Loader ve read_files() bunları doğrudan destekler. Formatı cloudFiles.format Python'da veya format => argümanı SQL'de belirtin. Kaynağı kontrol edebiliyorsanız, Parquet veya Avro’yu tercih edin; çünkü şema bilgisi içerirler ve daha iyi sıkıştırılırlar; bu da veri alımını ve şemanın çıkarılmasını hızlandırır. Boru hatları bu formatların herhangi birini yönetiyor, bu yüzden formatın kaynak seçiminizi kısıtlamasına izin vermeyin.
  • Ham depolama konumu: Dosyalar için, veri yönetilmemiş bir kova yolu yerine bir Unity Kataloğu hacmine yerleştirilir, böylece soy ve erişim kontrolü iniş bölgesine kadar uzanır. Bkz. Unity Catalog'un birimleri nelerdir?.

Bir boru hattının ürettiği tablolar için, kalan seçenekleriniz hedef katalog ve şemadır; bunlar yönetişim sınırını ve bulunabilirliğini belirler; ayrıca büyük tabloların fiziksel düzeni. Sorgu performansını, tablolar bölümleri manuel olarak ince ayar yapmadan büyürken iyi durumda tutmak için CLUSTER BY (sıvı kümeleme) kullanın. Bkz Tablolar için sıvı kümeleme kullanma.

Var olan bir tablodan yükleme

Azure Databricks'teki mevcut tablolardan veri yükleme. Sorgu kullanarak verileri dönüştürebilir veya işlem hattınızda daha fazla işlem için tabloyu yükleyebilirsiniz.

Piton

@dp.table(
  comment="A table summarizing counts of the top baby names for New York for 2021."
)
def top_baby_names_2021():
  return (
    spark.read.table("baby_names_prepared")
      .filter(expr("Year_Of_Birth == 2021"))
      .groupBy("First_Name")
      .agg(sum("Count").alias("Total_Count"))
      .sort(desc("Total_Count"))
  )

SQL

CREATE OR REFRESH MATERIALIZED VIEW top_baby_names_2021
COMMENT "A table summarizing counts of the top baby names for New York for 2021."
AS SELECT
  First_Name,
  SUM(Count) AS Total_Count
FROM baby_names_prepared
WHERE Year_Of_Birth = 2021
GROUP BY First_Name
ORDER BY Total_Count DESC

Dosyaları bulut nesne depolama alanından yükleme

Databricks, bulut nesne depolamasından veya Unity Kataloğu birimindeki dosyalardan veri alımı görevlerinin çoğu için işlem hatlarında Otomatik Yükleyici'nin kullanılmasını önerir. Auto Loader ve işlem hatları, sürekli büyüyen verileri bulut depolama alanına ulaşırken, artımlı ve değişmez bir şekilde yüklemek üzere tasarlanmıştır. Bkz. Otomatik Yükleyici nedir? ve Nesne depolamadan veri yükleme.

Aşağıdaki örnek, Otomatik Yükleyici'yi kullanarak bulut depolamadan verileri okur.

Piton

@dp.table
def customers():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .load("abfss://myContainer@myStorageAccount.dfs.core.windows.net/analysis/*/*/*.json")
  )

SQL

CREATE OR REFRESH STREAMING TABLE sales
  AS SELECT *
  FROM STREAM read_files(
    'abfss://myContainer@myStorageAccount.dfs.core.windows.net/analysis/*/*/*.json',
    format => "json"
  );

Aşağıdaki örneklerde, Unity Kataloğu birimindeki CSV dosyalarından veri kümeleri oluşturmak için Otomatik Yükleyici kullanılır.

Piton

@dp.table
def customers():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "csv")
      .load("/Volumes/my_catalog/retail_org/customers/")
  )

SQL

CREATE OR REFRESH STREAMING TABLE customers
AS SELECT * FROM STREAM read_files(
  "/Volumes/my_catalog/retail_org/customers/",
  format => "csv"
)

Uyarı

  • Dosya bildirimleriyle Otomatik Yükleyici'yi kullanır ve işlem hattınız veya akış tablonuz için tam yenileme çalıştırırsanız kaynaklarınızı el ile temizlemeniz gerekir. Temizleme gerçekleştirmek için not defterindeki CloudFilesResourceManager kullanabilirsiniz.
  • Unity Kataloğu etkin bir işlem hattında Otomatik Yükleyici ile dosya yüklemek için dış konumları kullanmanız gerekir. Unity Kataloğu'nu işlem hatlarıyla kullanma hakkında daha fazla bilgi edinmek için bkz. Unity Kataloğu'nu işlem hatlarıyla kullanma.

Bulut depolamada kimlik doğrulaması

Otomatik Yükleyici, bulut depolamada kimlik doğrulaması yapmak için Unity Kataloğu dış konumlarını kullanır. Okumak istediğiniz depolama yolu için bir dış konum yapılandırmanız ve yürüten kullanıcıya ayrıcalığı tanımanız READ FILES gerekir.

Azure Data Lake Storage'den veri almak için, bir depolama kapsayıcısını referans alan bir depolama kimlik bilgisi tarafından yedeklenen bir harici konum yapılandırın. Daha fazla bilgi için bkz. Unity Kataloğu'nu kullanarak bulut nesne depolamasına bağlanma.

İleti veri yolu'ndan veri yükleme

İleti veri hatlarından veri almak için işlem hatlarını yapılandırabilirsiniz. Databricks, ileti veri kümelerinden düşük gecikme süreli yükleme için en verimli alımı sağlamak üzere sürekli yürütme ve gelişmiş otomatik ölçeklendirme ile akış tablolarının kullanılmasını önerir. Daha fazla bilgi için bkz. Otomatik ölçeklendirme ile Lakeflow işlem hattı kümesi kullanımını iyileştirme.

Örneğin aşağıdaki kod, read_kafka işlevini kullanarak Kafka'dan veri almak için bir akış tablosu yapılandırır.

Piton

from pyspark import pipelines as dp

@dp.table
def kafka_raw():
  return (
    spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "kafka_server:9092")
      .option("subscribe", "topic1")
      .load()
  )

SQL

CREATE OR REFRESH STREAMING TABLE kafka_raw AS
  SELECT *
  FROM STREAM read_kafka(
    bootstrapServers => 'kafka_server:9092',
    subscribe => 'topic1'
  );

Google Pub/Sub'dan veri alma

Aşağıdaki örnek, read_pubsub işlevini kullanarak bir Google Pub/Sub başlığından okuyan bir akış tablosu oluşturur.

Piton

@dp.table
def pubsub_raw():
  auth_options = {
    "clientId": client_id,
    "clientEmail": client_email,
    "privateKey": private_key,
    "privateKeyId": private_key_id
  }
  return (
    spark.readStream
      .format("pubsub")
      .option("subscriptionId", "my-subscription")
      .option("topicId", "my-topic")
      .option("projectId", "my-project")
      .options(auth_options)
      .load()
  )

SQL

CREATE OR REFRESH STREAMING TABLE pubsub_raw
AS SELECT * FROM STREAM read_pubsub(
  subscriptionId => 'my-subscription',
  projectId => 'my-project',
  topicId => 'my-topic',
  clientEmail => secret('pubsub-scope', 'clientEmail'),
  clientId => secret('pubsub-scope', 'clientId'),
  privateKeyId => secret('pubsub-scope', 'privateKeyId'),
  privateKey => secret('pubsub-scope', 'privateKey')
);

Databricks, yetkilendirme seçenekleri sunarken gizli bilgilerin kullanılmasını önerir. Tüm kimlik doğrulama seçenekleri için bkz. Pub/Sub erişimini yapılandırma .

Diğer ileti veri yolu kaynaklarından almak için aşağıdaki bilgilere bakınız:

Azure Event Hubs den veri yükleme

Azure Event Hubs, Apache Kafka uyumlu bir arabirim sağlayan bir veri akışı hizmetidir. Azure Event Hubs'dan iletileri yüklemek için işlem hattı çalışma zamanına dahil olan Yapılandırılmış Akış Kafka bağlayıcısını kullanabilirsiniz. Azure Event Hubs'dan iletileri yükleme ve işleme hakkında daha fazla bilgi edinmek için bkz. Azure Event Hubs'ı işlem hattı veri kaynağı olarak kullanma.

Dış sistemlerden veri yükleme

İşlem hatları, Azure Databricks tarafından desteklenen herhangi bir veri kaynağından veri yüklemeyi destekler. Bkz . Veri kaynaklarına ve dış hizmetlere bağlanma. Dış verileri, veri kaynaklarınıntarafından desteklendiği durumlarda Lakehouse Federasyonu kullanarak da yükleyebilirsiniz. Lakehouse Federation, Databricks Runtime 13.3 LTS veya üzerini gerektirdiğinden, Lakehouse Federation'ı kullanmak için işlem hattınızı önizleme kanalını kullanacak şekilde yapılandırın.

Bazı veri kaynaklarının eşdeğer SQL desteği yoktur. Lakehouse Federasyon'unu bu veri kaynaklarından biriyle kullanamıyorsanız, kaynaktan veri almak için Python'ı kullanabilirsiniz. Python ve SQL kaynak dosyalarını aynı işlem hattına ekleyebilirsiniz. Aşağıdaki örnek, uzak bir PostgreSQL tablosundaki verilerin mevcut durumuna erişmek için maddileştirilmiş bir görünüm bildirir.

import dp

@dp.table
def postgres_raw():
  return (
    spark.read
      .format("postgresql")
      .option("dbtable", table_name)
      .option("host", database_host_url)
      .option("port", 5432)
      .option("database", database_name)
      .option("user", username)
      .option("password", password)
      .load()
  )

Bulut nesne depolama alanından küçük veya statik veri kümeleri yükleme

Apache Spark yük söz dizimini kullanarak küçük veya statik veri kümelerini yükleyebilirsiniz. İşlem hatları, Azure Databricks apache Spark tarafından desteklenen tüm dosya biçimlerini destekler. Tam liste için bkz. Veri biçimi seçenekleri.

Aşağıdaki örneklerde, tablo oluşturmak için JSON yükleme gösterilmektedir.

Piton

@dp.table
def clickstream_raw():
  return (spark.read.format("json").load("/databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed-json/2015_2_clickstream.json"))

SQL

CREATE OR REFRESH MATERIALIZED VIEW clickstream_raw
AS SELECT * FROM read_files(
  "/databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed-json/2015_2_clickstream.json"
)

Uyarı

read_files SQL işlevi, Azure Databricks'te tüm SQL ortamları için ortaktır. İşlem hatlarında SQL kullanarak doğrudan dosya erişimi için önerilen desendir. Daha fazla bilgi için bkz. Seçenekler.

Python özel veri kaynağından veri yükleme

Python özel veri kaynakları, verileri özel biçimlerde yüklemenize olanak sağlar. Belirli bir dış veri kaynağından okumak ve bu kaynağa yazmak için kod yazabilir veya kendi iç sistemlerinizdeki verileri okumak için mevcut Python kodunuzu kullanabilirsiniz. Python veri kaynakları geliştirme hakkında daha fazla ayrıntı için bkz. PySpark özel veri kaynakları.

Aşağıdaki örnek, özel bir veri kaynağını biçim adıyla my_custom_datasource kaydeder ve hem toplu işlem hem de akış modlarında bu kaynaktan okur.

from pyspark import pipelines as dp

# Assume `my_custom_datasource` is a custom Python custom data
# source that supports both batch and streaming reads, and has
# been registered using `spark.dataSource.register`.

# This creates a materialized view
@dp.table(name = "read_from_batch")
def read_from_batch():
    return spark.read.format("my_custom_datasource").load()

# This creates a streaming table
@dp.table(name = "read_from_streaming")
def read_from_streaming():
    return spark.readStream.format("my_custom_datasource").load()

Kaynak akış tablosundaki değişiklikleri yoksayacak şekilde bir akış tablosunu yapılandırma

Varsayılan olarak, akış tabloları yalnızca ekleme kaynakları gerektirir. Kaynak akış tablonuz güncelleme veya silme işlemleri gerektiriyorsa (örneğin, GDPR “unutulma hakkı” işleme süreci için), bu değişiklikleri yok saymak üzere skipChangeCommits bayrağını kullanın. Bu bayrak yalnızca spark.readStream işlevini kullanarak option() çalışır ve kaynak akış tablosu bir create_auto_cdc_flow() işlevinin hedefi olduğunda kullanılamaz. Daha fazla bilgi için bkz. Kaynak Delta Lake tablolarındaki değişiklikleri işleme.

@dp.table
def b():
   return spark.readStream.option("skipChangeCommits", "true").table("A")

İşlem hattında gizli bilgilerle depolama kimlik bilgilerine güvenli bir şekilde erişme

Erişim anahtarları veya parolalar gibi kimlik bilgilerini depolamak için Azure Databricks gizli dizileri kullanabilirsiniz. Gizliyi işlem hattınıza yapılandırmak için işlem hattı ayarları küme yapılandırmasında bir Spark özelliği kullanın. Boru hatları için klasik işlem yapılandırmasını yapılandırbkz.

Aşağıdaki örnek, Auto Loader kullanarak bir Azure Data Lake Storage depolama hesabından veri okumak için gereken erişim anahtarını depolamak amacıyla bir gizli anahtar kullanır. İşlem hattınızın gerektirdiği gizli dizileri (örneğin, S3'e erişmek için AWS anahtarları veya apache Hive meta veri deposu parolası) yapılandırmak için aynı yöntemi kullanabilirsiniz.

Azure Data Lake Storage ile çalışma hakkında daha fazla bilgi edinmek için bkz. Azure Data Lake Storage'a bağlanma ve Blob Depolama.

Uyarı

gizli dizi değerini ayarlayan spark.hadoop. yapılandırma anahtarına spark_conf ön ekini eklemeniz gerekir.

{
  "id": "43246596-a63f-11ec-b909-0242ac120002",
  "storage": "abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/<path>",
  "clusters": [
    {
      "spark_conf": {
        "spark.hadoop.fs.azure.account.key.<storage-account-name>.dfs.core.windows.net": "{{secrets/<scope-name>/<secret-name>}}"
      },
      "autoscale": {
        "min_workers": 1,
        "max_workers": 5,
        "mode": "ENHANCED"
      }
    }
  ],
  "development": true,
  "continuous": false,
  "libraries": [
    {
      "notebook": {
        "path": "/Users/user@databricks.com/Pipeline Notebooks/pipeline quickstart"
      }
    }
  ],
  "name": "pipeline quickstart using ADLS2"
}

Bu kod örneğinde aşağıdaki değerleri değiştirin.

Yer tutucu bununla değiştir
<container-name> Azure depolama hesabı kapsayıcısının adı.
<storage-account-name> ADLS depolama hesabı adı.
<path> İşlem hattı çıktı verilerinin ve meta verilerinin yolu.
<scope-name> Azure Databricks gizli kapsam adı.
<secret-name> Azure depolama hesabı erişim anahtarını içeren anahtarın adı. 
from pyspark import pipelines as dp

json_path = "abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/<path-to-input-dataset>"
@dp.create_table(
  comment="Data ingested from an ADLS2 storage account."
)
def read_from_ADLS2():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .load(json_path)
  )

Bu kod örneğinde aşağıdaki değerleri değiştirin.

Yer tutucu bununla değiştir
<container-name> Giriş verilerini depolayan Azure depolama hesabı kapsayıcısının adı.
<storage-account-name> ADLS depolama hesabı adı.
<path-to-input-dataset> Giriş veri kümesinin yolu.
  • Last updated on