Memuat data dalam alur

Anda dapat memuat data dari sumber data apa pun yang didukung oleh Apache Spark di Azure Databricks menggunakan alur. Anda dapat mendefinisikan dataset (tabel dan tampilan) dalam pipeline berdasarkan kueri apa pun yang menghasilkan Spark DataFrame, termasuk streaming DataFrame dan Pandas for Spark DataFrame. Untuk tugas penyerapan data, Databricks merekomendasikan penggunaan tabel streaming untuk sebagian besar kasus penggunaan. Tabel streaming berguna untuk menyerap data dari penyimpanan objek cloud menggunakan Auto Loader atau dari bus pesan seperti Kafka. Untuk informasi selengkapnya tentang tabel streaming, jenis himpunan data utama untuk penyerapan, lihat Tabel streaming.

Tidak semua sumber data memiliki dukungan SQL untuk penyerapan. Namun, Anda dapat mencampur sumber SQL dan Python dalam alur yang sama untuk menggunakan Python jika diperlukan. Untuk detail selengkapnya tentang menggunakan pustaka yang secara default tidak disertakan dalam pipeline, lihat Mengelola dependensi Python untuk pipeline. Untuk informasi umum tentang penyerapan di Azure Databricks, lihat Konektor standar di Lakeflow Connect.

Contoh berikut menunjukkan beberapa pola pemuatan data umum.

Memuat dari tabel yang sudah ada

Muat data dari tabel yang sudah ada di Azure Databricks. Anda dapat mengubah data menggunakan kueri atau memuat tabel untuk pemrosesan lebih lanjut di alur Anda.

Phyton

@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

Memuat file dari penyimpanan objek cloud

Databricks merekomendasikan penggunaan Auto Loader dalam pipeline untuk sebagian besar tugas pengambilan data dari penyimpanan objek cloud atau dari file dalam volume Katalog Unity. Auto Loader dan pipeline dirancang untuk memuat data yang terus bertambah secara inkremental dan idempoten saat tiba di penyimpanan cloud. Lihat Apa itu Auto Loader? dan Memuat data dari penyimpanan objek.

Contoh berikut membaca data dari penyimpanan cloud menggunakan Auto Loader.

Phyton

@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"
  );

Contoh berikut menggunakan Auto Loader untuk membuat himpunan data dari file CSV dalam volume Katalog Unity.

Phyton

@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"
)

Nota

  • Jika Anda menggunakan Auto Loader dengan pemberitahuan file dan menjalankan refresh penuh untuk alur atau tabel streaming, Anda harus membersihkan sumber daya Anda secara manual. Anda dapat menggunakan CloudFilesResourceManager di notebook untuk melakukan pembersihan.
  • Untuk memuat file dengan Auto Loader dalam alur Katalog Unity yang diaktifkan, Anda harus menggunakan lokasi eksternal. Untuk mempelajari selengkapnya tentang menggunakan Katalog Unity dengan alur, lihat Menggunakan Katalog Unity dengan alur.

Mengautentikasi ke penyimpanan cloud

Auto Loader menggunakan lokasi eksternal Katalog Unity untuk mengautentikasi terhadap penyimpanan cloud. Anda harus mengonfigurasi lokasi eksternal untuk jalur penyimpanan yang ingin Anda baca dan memberikan READ FILES hak istimewa kepada pengguna yang mengeksekusi.

Untuk menyerap dari Azure Data Lake Storage, konfigurasikan lokasi eksternal yang didukung oleh kredensial penyimpanan yang mereferensikan kontainer penyimpanan. Untuk informasi selengkapnya, lihat Menyambungkan ke penyimpanan objek cloud menggunakan Katalog Unity.

Memuat data dari bus pesan

Anda dapat mengonfigurasi alur untuk menyerap data dari bus pesan. Databricks merekomendasikan penggunaan tabel streaming dengan eksekusi berkelanjutan dan penskalaan otomatis yang lebih canggih untuk memberikan penyerapan paling efisien untuk beban latensi rendah dari bus pesan. Untuk informasi selengkapnya, lihat Mengoptimalkan pemanfaatan kluster alur Lakeflow dengan penskalaan otomatis.

Misalnya, kode berikut mengonfigurasi tabel streaming untuk menyerap data dari Kafka menggunakan fungsi read_kafka .

Phyton

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'
  );

Impor dari Google Pub/Sub

Contoh berikut membuat tabel streaming yang dibaca dari topik Google Pub/Sub menggunakan fungsi read_pubsub .

Phyton

@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 merekomendasikan penggunaan rahasia saat memberikan opsi otorisasi. Lihat Mengonfigurasi akses ke Pub/Sub untuk semua opsi autentikasi.

Untuk mengambil data dari sumber bus pesan lainnya, lihat:

Load data dari Azure Event Hubs

Azure Event Hubs adalah layanan streaming data yang menyediakan antarmuka yang kompatibel dengan Apache Kafka. Anda dapat menggunakan konektor Kafka untuk Structured Streaming, yang disertakan dalam runtime pipeline, untuk memuat pesan dari Azure Event Hubs. Untuk mempelajari selengkapnya tentang memuat dan memproses pesan dari Azure Event Hubs, lihat Menggunakan Azure Event Hubs sebagai sumber data alur.

Memuat data dari sistem eksternal

Alur mendukung pemuatan data dari sumber data apa pun yang didukung oleh Azure Databricks. Lihat Menyambungkan ke sumber data dan layanan eksternal. Anda juga dapat memuat data eksternal menggunakan Federasi Lakehouse untuk sumber data yang didukung. Karena Lakehouse Federation memerlukan Databricks Runtime 13.3 LTS atau lebih tinggi, untuk menggunakan Lakehouse Federation, konfigurasikan alur Anda untuk menggunakan saluran pratinjau.

Beberapa sumber data tidak memiliki dukungan SQL yang setara. Jika Anda tidak dapat menggunakan Federasi Lakehouse dengan salah satu sumber data ini, Anda dapat menggunakan Python untuk menyerap data dari sumbernya. Anda dapat menambahkan file sumber Python dan SQL ke alur yang sama. Contoh berikut mendeklarasikan tampilan materialisasi untuk mengakses status data saat ini dalam tabel PostgreSQL jarak jauh.

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()
  )

Memuat himpunan data kecil atau statis dari penyimpanan objek cloud

Anda dapat memuat himpunan data kecil atau statis menggunakan sintaks beban Apache Spark. Alur mendukung semua format file yang didukung oleh Apache Spark pada Azure Databricks. Untuk daftar lengkapnya, lihat Opsi format data .

Contoh berikut menunjukkan pemuatan JSON untuk membuat tabel.

Phyton

@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"
)

Nota

Fungsi SQL read_files umum untuk semua lingkungan SQL di Azure Databricks. Ini adalah pola yang direkomendasikan untuk akses file langsung menggunakan SQL dalam alur. Untuk informasi selengkapnya, lihat opsi .

Memuat data dari sumber data kustom Python

Sumber data kustom Python memungkinkan Anda memuat data dalam format kustom. Anda dapat menulis kode untuk membaca dan menulis ke sumber data eksternal tertentu atau menggunakan kode Python yang ada untuk membaca data dari sistem internal Anda sendiri. Untuk detail selengkapnya tentang mengembangkan sumber data Python, lihat sumber data kustom PySpark.

Contoh berikut mendaftarkan sumber data kustom dengan nama my_custom_datasource format dan membacanya dalam mode batch dan streaming.

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()

Mengonfigurasi tabel streaming untuk mengabaikan perubahan dalam tabel streaming sumber

Secara default, tabel streaming memerlukan sumber khusus tambahan. Jika tabel streaming sumber Anda memerlukan pembaruan atau penghapusan (misalnya, untuk pemrosesan GDPR “hak untuk dilupakan”), gunakan flag skipChangeCommits untuk mengabaikan perubahan tersebut. Bendera ini hanya berfungsi dengan spark.readStream menggunakan option() fungsi dan tidak dapat digunakan ketika tabel streaming sumber adalah target fungsi create_auto_cdc_flow(). Untuk informasi selengkapnya, lihat Menangani perubahan pada tabel Delta Lake sumber.

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

Akses kredensial penyimpanan dengan aman menggunakan rahasia dalam alur kerja

Anda dapat menggunakan Azure Databricks rahasia untuk menyimpan kredensial seperti kunci akses atau kata sandi. Untuk mengonfigurasi kata sandi dalam alur Anda, gunakan properti Spark dalam konfigurasi kluster pengaturan alur. Lihat Mengonfigurasi komputasi klasik untuk alur.

Contoh berikut menggunakan rahasia untuk menyimpan kunci akses yang diperlukan untuk membaca data input dari akun penyimpanan Azure Data Lake Storage menggunakan Auto Loader. Anda dapat menggunakan metode yang sama ini untuk mengonfigurasi rahasia apa pun yang diperlukan oleh alur Anda, misalnya, kunci AWS untuk mengakses S3, atau kata sandi ke metastore Apache Hive.

Untuk informasi lebih lanjut tentang bekerja dengan Azure Data Lake Storage, lihat Menghubungkan ke Azure Data Lake Storage dan Blob Storage.

Nota

Anda harus menambahkan awalan spark.hadoop. ke kunci konfigurasi spark_conf yang mengatur nilai rahasia.

{
  "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"
}

Dalam sampel kode ini, ganti nilai berikut.

Placeholder Ganti dengan
<container-name> Nama kontainer akun penyimpanan Azure.
<storage-account-name> Nama akun penyimpanan ADLS.
<path> Jalur untuk data keluaran pengaliran dan metadata.
<scope-name> Nama cakupan rahasia Azure Databricks.
<secret-name> Nama kunci yang berisi kunci akses akun penyimpanan Azure. 
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)
  )

Dalam sampel kode ini, ganti nilai berikut.

Placeholder Ganti dengan
<container-name> Nama kontainer akun penyimpanan Azure yang menyimpan data input.
<storage-account-name> Nama akun penyimpanan ADLS.
<path-to-input-dataset> Jalur ke himpunan data input.
  • Last updated on