パイプラインにデータを読み込む

パイプラインを使用して、Azure Databricks 上の Apache Spark でサポートされている任意のデータ ソースからデータを読み込むことができます。 Spark DataFrame を返すクエリ (ストリーミング DataFrames や Spark DataFrames 用 Pandas など) に対して、パイプライン内のデータセット (テーブルとビュー) を定義できます。 データ インジェスト タスクの場合、Databricks では、ほとんどのユース ケースでストリーミング テーブルを使用することをお勧めします。 ストリーミング テーブルは、自動ローダーを使用してクラウド オブジェクト ストレージから、または Kafka などのメッセージ バスからデータを取り込む場合に便利です。 インジェストのプライマリ データセットの種類であるストリーミング テーブルの詳細については、「 ストリーミング テーブル」を参照してください。

すべてのデータ ソースで、インジェストに対して SQL がサポートされているわけではありません。 ただし、SQL ソースとPython ソースを同じパイプラインに混在させ、必要に応じてPythonを使用できます。 既定でパイプラインにパッケージ化されていないライブラリの操作の詳細については、「パイプラインのPython依存関係の管理」を参照してください。 Azure Databricks でのインジェストの一般的な情報については、 Lakeflow Connect の Standard コネクタに関するページを参照してください。

次の例では、一般的なデータ読み込みパターンを示します。

データソースと接続経路を特定しましょう

パイプラインコードを書く前に、データの出所をすべて在庫に入れてください。 各ソースについて、そのデータ(ファイル、データベース、SaaS(ソフトウェア・アズ・ア・サービス)、API、ストリームなど、どのようにデータを公開しているか、どのくらいの頻度で変更されるか、必要な認証情報やネットワークアクセスについて注目してください。 接続方式はソースが自然にバッチ処理かストリーミングかを判断することが多いため、早い段階で正しく調整することで後々のやり直しを避けられます。

各ソースを以下のいずれかの接続パスにバケット分けします。 以下の表は、各パイプラインソースの推奨メカニズムを示しています。

各ソースについて、構築前に以下の点を確認してください:

• ID: パイプラインがどの ID で実行されるか。 パイプラインはサービスプリンシパルとして動作できるので、個人アカウントに頼らないようにまず設定してください。
• ネットワークパス:ソースが必要とする接続性、例えばストレージ認証情報、外部ロケーション、またはLakeflow Connectの管理接続性など。
• 意味の変更:ソースが更新や削除を通知する際の方法(もしある場合)。 これにより、CDCが必要か、ソースを付録のみ扱うかが決まります。

ファイル形式とストレージレイヤーを選択します

パイプラインがこの決定の大部分を代わりに行います。 パイプラインが作成するすべてのストリーミングテーブルやマテリアライズされたビューはデフォルトでDeltaテーブルとして保存されており、これによりACIDトランザクション、スキーマの強制と進化、タイムトラベル、Unityカタログのガバナンスや系譜がすべてのデータセットで提供されます。 パイプライン出力のフォーマットは選べません。 あなたの本当の決断はパイプラインの両端にあります:

• 生の入力フォーマット:CSV、JSON、Parquetなどソースが生成する形式です。 Auto Loaderと read_files() はこれらを直接サポートしています。 PythonではcloudFiles.format、SQLではformat =>引数でフォーマットを指定します。 ソースをコントロールするならParquetかAvroをおすすめします。これらはスキーマを扱いやすく圧縮も良く、取り込みやスキーマ推論が速くなります。 パイプラインはこれらの形式のいずれにも対応しているため、形式によってソースの選択肢を狭める必要はありません。
• 生データの保存場所: ファイルの場合は、ガバナンス対象外のバケットパスではなく Unity Catalog のボリュームにデータを配置することで、データリネージとアクセス制御をランディングゾーンにまで適用できます。 「Unity カタログ ボリュームとは」を参照してください。

パイプラインが生成するテーブルについては、残りの選択肢はターゲットカタログとスキーマ(ガバナンスの境界と発見可能性を設定する)、そして大きなテーブルの物理レイアウトです。 CLUSTER BY(リキッドクラスタリング)を使って、パーティションを手動調整せずにテーブルが成長する際のクエリ性能を良好に保ちましょう。 表に液体クラスタリングを使用するを参照してください。

既存のテーブルからの読み込み

Azure Databricks 内の既存のテーブルからデータを読み込みます。 クエリを使用してデータを変換したり、パイプラインでさらに処理するためにテーブルを読み込んだりすることができます。

Python

@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

クラウド オブジェクト ストレージからファイルを読み込む

Databricks では、クラウド オブジェクト ストレージまたは Unity カタログ ボリューム内のファイルからのほとんどのデータ インジェスト タスクに対して、パイプラインで自動ローダーを使用することをお勧めします。 Auto Loader とパイプラインは、クラウド ストレージに到着する増加し続けるデータを、増分的かつべき等性を持って読み込むように設計されています。 「自動ローダーとは」を参照し、オブジェクト ストレージからデータを読み込みます。

次の例では、自動ローダーを使用してクラウド ストレージからデータを読み取ります。

Python

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

次の例では、自動ローダーを使用して、Unity カタログ ボリューム内の CSV ファイルからデータセットを作成します。

Python

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

クラウド ストレージに対する認証

自動ローダーは、Unity カタログの外部の場所を使用して、クラウド ストレージに対する認証を行います。 読み取るストレージ パスの外部の場所を構成し、実行中のユーザーに READ FILES 特権を付与する必要があります。

Azure Data Lake Storageから取り込むには、ストレージ コンテナーを参照するストレージ資格情報に基づく外部の場所を構成します。 詳細については、「 Unity カタログを使用してクラウド オブジェクト ストレージに接続する」を参照してください。

メッセージ バスからデータを読み込む

メッセージ バスからデータを取り込むパイプラインを構成できます。 Databricks では、メッセージ バスからの待機時間の短い読み込みに最も効率的なインジェストを提供するために、継続的な実行と拡張された自動スケーリングでストリーミング テーブルを使用することをお勧めします。 詳細については、「 自動スケーリングを使用した Lakeflow パイプライン クラスターの使用率の最適化」を参照してください。

たとえば、次のコードは、 read_kafka 関数を使用して Kafka からデータを取り込むためのストリーミング テーブルを構成します。

Python

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 から取り込む

次の例では、 read_pubsub 関数を使用して Google Pub/Sub トピックから読み取るストリーミング テーブルを作成します。

Python

@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 では、承認オプションを提供するときにシークレットを使用することをお勧めします。 すべての認証オプションについては、 Pub/Sub へのアクセスの構成 を参照してください。

他のメッセージ バス ソースから取り込むには、次を参照してください。

• キネシス: read_kinesis
• パルサー: read_pulsar

Azure Event Hubs からデータを読み込む

Azure Event Hubs は、Apache Kafka 互換インターフェイスを提供するデータ ストリーミング サービスです。 パイプライン ランタイムに含まれる Structured Streaming Kafka コネクタを使用して、Azure Event Hubsからメッセージを読み込むことができます。 Azure Event Hubs からのメッセージの読み込みと処理の詳細については、「 Azure Event Hubs をパイプライン データ ソースとして使用する」を参照してください。

外部システムからデータを読み込む

パイプラインでは、Azure Databricksでサポートされている任意のデータ ソースからのデータの読み込みがサポートされます。 データ ソースと外部サービスへの接続を参照してください。 サポートされているデータ ソースに対して Lakehouse フェデレーションを使用して外部データを読み込むこともできます。 Lakehouse Federation では Databricks Runtime 13.3 LTS 以上が必要であるため、Lakehouse Federation を使用するには、 プレビュー チャネルを使用するようにパイプラインを構成します。

一部のデータ ソースには、同等の SQL サポートがありません。 これらのデータ ソースのいずれかで Lakehouse Federation を使用できない場合は、Python を使用してソースからデータを取り込むことができます。 Python と SQL のソース ファイルを同じパイプラインに追加できます。 次の例では、リモート PostgreSQL テーブル内のデータの現在の状態にアクセスするための具体化されたビューを宣言します。

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

クラウド オブジェクト ストレージから小さいデータセットまたは静的データセットを読み込む

Apache Spark の読み込み構文を使用して、小規模または静的なデータセットを読み込むことができます。 パイプラインは、Azure Databricks上の Apache Spark でサポートされているすべてのファイル形式をサポートします。 完全な一覧については、「 データ形式のオプション」を参照してください。

次の例では、JSON を読み込んでテーブルを作成する方法を示します。

Python

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

Python カスタム データ ソースからデータを読み込む

Python カスタム データ ソースを使用すると、カスタム形式でデータを読み込むことができます。 特定の外部データ ソースの読み取りと書き込みを行うコードを記述することも、既存のPython コードを使用して独自の内部システムからデータを読み取ることもできます。 Python データ ソースの開発の詳細については、「 PySpark カスタム データ ソース」を参照してください。

次の例では、カスタム データ ソースを my_custom_datasource 形式名で登録し、バッチ モードとストリーミング モードの両方で読み取ります。

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

ソース ストリーミング テーブルの変更を無視するようにストリーミング テーブルを構成する

既定では、ストリーミング テーブルには追加専用のソースが必要です。 ソース ストリーミング テーブルで更新または削除が必要な場合 (たとえば、GDPR の "忘れられる権利" 処理の場合)、 skipChangeCommits フラグを使用してそれらの変更を無視します。 このフラグは、spark.readStream関数を使用するoption()でのみ機能し、ソース ストリーミング テーブルが create_auto_cdc_flow() 関数のターゲットである場合は使用できません。 詳細については、「 ソース Delta Lake テーブルへの変更を処理する」を参照してください。

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

パイプライン内のシークレットを使用してストレージ資格情報に安全にアクセスする

Azure Databricks シークレットを使用して 、 アクセス キーやパスワードなどの資格情報を格納できます。 パイプラインでシークレットを構成するには、パイプライン設定クラスター構成で Spark プロパティを使用します。 パイプラインのクラシック コンピューティングの構成に関するページを参照してください。

次の例では、シークレットを使用して、自動ローダーを使用してAzure Data Lake Storageストレージ アカウントから入力データを読み取るために必要なアクセス キーを格納します。 この同じ方法を使用して、パイプラインに必要なシークレット (たとえば、S3 にアクセスするための AWS キー、Apache Hive メタストアへのパスワードなど) を構成できます。

Azure Data Lake Storage の操作の詳細については、Azure Data Lake Storage と Blob Storage への接続に関するページを参照してください。

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

このコード サンプルでは、次の値を置き換えます。

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

このコード サンプルでは、次の値を置き換えます。