メインコンテンツまでスキップ
に最終更新

パイプラインでデータをロードする

パイプラインを使用して、Databricks の Apache Spark でサポートされている任意のデータソースからデータをロードできます。パイプラインでは、ストリーミング DataFrames および Spark DataFrames 用の Pandas を含む、Spark DataFrame を返す任意のクエリーに対してデータセット (テーブルとビュー) を定義できます。データ取り込みタスクの場合、Databricks はほとんどのユースケースでストリーミングテーブルの使用を推奨しています。ストリーミングテーブルは、Auto Loader を使用してクラウドオブジェクトストレージからデータを、または Kafka などのメッセージバスからデータを取り込むのに役立ちます。取り込みの主要なデータセットタイプであるストリーミングテーブルの詳細については、「ストリーミングテーブル」を参照してください。

すべてのデータソースが取り込み用のSQLをサポートしているわけではありません。ただし、同じパイプライン内でSQLソースとPythonソースを混在させ、必要な箇所でPythonを使用することができます。defaultでパイプラインにパッケージ化されていないライブラリの操作の詳細については、「パイプラインのPython依存関係の管理」を参照してください。Databricksでの取り込みに関する一般的な情報については、「標準コネクタの選択」を参照してください。

以下の例は、一般的なデータ読み込みパターンを示しています。

データソースと接続パスを特定する​

パイプラインコードを作成する前に、データのすべての発生元をリストアップしてください。各ソースについて、データがどのように公開されているか（ファイル、データベース、SaaS システム、API、またはストリーム）、変更頻度、および必要な資格情報とネットワークアクセスを記録してください。接続方法は、ソースが本質的にバッチであるかストリーミングであるかを決定することが多いため、早い段階で正しく設定しておくことで、後の手戻りを防ぐことができます。

各ソースを以下の接続パスのいずれかに分類します。次の表に、各パイプラインソースの推奨メカニズムを記載します。

ソース

接続パス

クラウドオブジェクトストレージ (S3、Azure Data Lake Storage、GCS) に格納されるファイル

最も一般的な開始点です。インクリメンタルな検出、スキーマ推論、およびスキーマ進化を処理する Auto Loader (cloudFiles 形式) を使用します。クラウドオブジェクトストレージからのファイルの読み込みを参照してください。

データベースおよびSaaSアプリケーション(Salesforce、SQL Server、PostgreSQL、Workday)

ソースで利用可能な場合は、Lakeflow Connectのマネージドコネクタを使用してください。マネージド コネクタは構成主導型であり、認証や増分抽出、またはCDC抽出を自動的に処理します。Lakeflow Connectのコネクタの概念を参照してください。ソース用のマネージド コネクタが存在しない場合は、直接取り込むか、応答を最初にファイルとして保存してください。「パイプラインでのAPIからのデータの取り込み」を参照してください。

メッセージバス (Kafka、Kinesis、Azure Event Hubs、Pub/Sub)

これらはネイティブストリーミングソースなので、直接Structured Streamingソースとして読み取ってください。メッセージバスからデータをロードするを参照してください。

その他の Delta テーブルまたは Unity Catalog アセット（他のパイプラインやジョブによって生成されたテーブルを含む）

それらを直接参照し、Unity Catalogのガバナンスとリネージに発見とアクセスを任せましょう。既存のテーブルから読み込むを参照してください。

小規模または静的な参照データ (ルックアップファイル、ほとんど変更されない CSV)

マテリアライズドビューにバッチソースとしてロードします。ほとんど変わらないものをストリーミングしてもメリットはありません。「 クラウドオブジェクトストレージから小規模または静的データセットをロードする」を参照してください。

管理されたコネクターを持たない任意のHTTPまたはREST API

パイプライン内のAPIから取得するか、まずそのレスポンスをファイルとして到達させてください。パイプラインでAPIからデータを取り込む方法については、こちらをご覧ください。

ソース

接続パス

クラウドオブジェクトストレージ (S3、Azure Data Lake Storage、GCS) に格納されるファイル

最も一般的な開始点です。インクリメンタルな検出、スキーマ推論、およびスキーマ進化を処理する Auto Loader (cloudFiles 形式) を使用します。クラウドオブジェクトストレージからのファイルの読み込みを参照してください。

データベースおよびSaaSアプリケーション(Salesforce、SQL Server、PostgreSQL、Workday)

ソースで利用可能な場合は、Lakeflow Connectのマネージドコネクタを使用してください。マネージド コネクタは構成主導型であり、認証や増分抽出、またはCDC抽出を自動的に処理します。Lakeflow Connectのコネクタの概念を参照してください。ソース用のマネージド コネクタが存在しない場合は、直接取り込むか、応答を最初にファイルとして保存してください。「パイプラインでのAPIからのデータの取り込み」を参照してください。

メッセージバス (Kafka、Kinesis、Azure Event Hubs、Pub/Sub)

これらはネイティブストリーミングソースなので、直接Structured Streamingソースとして読み取ってください。メッセージバスからデータをロードするを参照してください。

その他の Delta テーブルまたは Unity Catalog アセット（他のパイプラインやジョブによって生成されたテーブルを含む）

それらを直接参照し、Unity Catalogのガバナンスとリネージに発見とアクセスを任せましょう。既存のテーブルから読み込むを参照してください。

小規模または静的な参照データ (ルックアップファイル、ほとんど変更されない CSV)

マテリアライズドビューにバッチソースとしてロードします。ほとんど変わらないものをストリーミングしてもメリットはありません。「 クラウドオブジェクトストレージから小規模または静的データセットをロードする」を参照してください。

管理されたコネクターを持たない任意のHTTPまたはREST API

パイプライン内のAPIから取得するか、まずそのレスポンスをファイルとして到達させてください。パイプラインでAPIからデータを取り込む方法については、こちらをご覧ください。

各ソースについて、ビルド前に次のことを確認してください：

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

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

パイプラインが、この決定の大部分を自動的に行います。パイプラインによって作成されるすべてのストリーミングテーブルとマテリアライズドビューは、defaultでDeltaテーブルとして保存されます。これにより、すべてのデータセットに対して、ACIDトランザクション、スキーマ強制と進化、タイムトラベル、およびUnity Catalogによるガバナンスとリネージが提供されます。パイプライン出力の形式を選択することはできません。実際の意思決定は、パイプラインの両端で行われます：

  • 生の入力フォーマット : CSV、JSON、Parquetなど、ソースが生成するあらゆるもの。Auto Loaderおよびread_files()は、これらを直接サポートしています。PythonではcloudFiles.formatを使用して、SQLではformat =>引数を使用してフォーマットを指定します。ソースを制御できる場合は、ParquetまたはAvroの使用を推奨します。これらはスキーマを保持し、圧縮効率も高いため、取り込みとスキーマ推論が高速化されます。パイプラインはこれらのフォーマットのいずれも処理できるため、フォーマットによってソースの選択が制限されないようにしてください。
  • 生のストレージ場所 :ファイルの場合、ランドデータはUnovernedバケットパスではなくUnity Catalogボリュームに配置し、リネージとアクセス制御はランディングゾーンまで遡ります。See What are Unity Catalog volumes?.

パイプラインが生成するテーブルについて、残りの選択肢はターゲットカタログとスキーマです。これらはガバナンスの境界と検出可能性、および大規模テーブルの物理Layoutを設定します。パーティションを手動で調整することなく、テーブルの増加に伴うクエリーパフォーマンスを良好に保つには、CLUSTER BY（リキッドクラスタリング）を使用してください。テーブルにリキッドクラスタリングを使用するを参照してください。

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

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

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

Databricks 、クラウド オブジェクト ストレージまたはUnity Catalogボリューム内のファイルからのデータ取り込みタスクのほとんどに、パイプラインのAuto Loader使用することをお勧めします。 Auto Loaderとパイプラインは、増え続けるデータがクラウド ストレージに到着すると、増分的かつべき等にロードするように設計されています。 Auto Loaderとは何か？」および「オブジェクトストレージからデータをロードする」を参照してください。

以下の例では、Auto Loader を使用してクラウドストレージからデータを読み取ります。

Python
@dp.table
def customers():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .load("gs://mybucket/analysis/*/*/*.json")
  )

次の例では、 Auto Loader使用して、 Unity Catalogボリューム内のCSVファイルからデータセットを作成します。

Python
@dp.table
def customers():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "csv")
      .load("/Volumes/my_catalog/retail_org/customers/")
  )
注記
  • ファイル通知でAuto Loaderを使用し、パイプラインまたはストリーミング テーブルの完全な更新を実行する場合は、リソースを手動でクリーンアップする必要があります。 ノートブック内のCloudFilesResourceManagerを使用してクリーンアップを実行できます。
  • Unity Catalog有効なパイプラインでAuto Loaderを使用してファイルをロードするには、外部ロケーションを使用する必要があります。 パイプラインで Unity Catalog を使用する方法の詳細については、 「パイプラインで Unity Catalog を使用する」を参照してください。

クラウドストレージへの認証​

Auto Loader Unity Catalog外部ロケーションを使用してクラウド ストレージに対して認証します。 読み取り先のストレージ パスの外部ロケーションを構成し、実行ユーザーにREAD FILES権限を付与する必要があります。

Google Cloud Storage (GCS) からデータを取り込むには、GCS バケットを参照するストレージ認証情報によって裏付けられた外部ロケーションを設定します。詳細については、 Google Cloud Storage ( GCS ) 外部ロケーションへの接続」を参照してください。

GCSでAuto Loaderのファイル通知モードを使用する必要がある場合は、 Auto LoaderオプションでDatabricksサービスの資格情報または Google サービス アカウントの資格情報を指定します。 Google固有の認証オプションについては、 GCPを参照してください。

次の例では、サービス資格情報を使用して、 GCSソースのファイル通知モードを有効にします。

Python
@dp.table
def customers():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .option("cloudFiles.useNotifications", "true")
      .option("cloudFiles.projectId", "my-gcp-project")
      .option("databricks.serviceCredential", "my-service-credential")
      .load("gs://my-bucket/path/to/files")
  )

ディレクトリ一覧表示モード（デフォルト）の場合、認証は外部ロケーションを通じて処理されるため、コード内に追加の認証情報は必要ありません。

メッセージバスからデータをロードする​

メッセージバスからデータを取り込むようにパイプラインを構成できます。Databricksでは、ストリーミングテーブルを連続実行と強化されたオートスケールと共に使用して、メッセージバスからの低遅延読み込みのための最も効率的な取り込みを提供することをお勧めします。詳細については、オートスケールを使用したLakeFlow Pipelinesクラスターの利用率を最適化するを参照してください。

例えば、以下のコードは、 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()
  )

Google Pub/Sub から取り込む​

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

Python
@dp.table
def pubsub_raw():
  return (
    spark.readStream
      .format("pubsub")
      .option("subscriptionId", "my-subscription")
      .option("topicId", "my-topic")
      .option("projectId", "my-project")
      .option("serviceCredential", "my-service-credential")
      .load()
  )

Databricksは、Pub/Sub認証にサービス認証情報を使用することを推奨しています。Pub/Sub のサービス認証情報には、Databricks Runtime 16.1 以降が必要です。サービス資格情報の作成を参照してください。

サービス認証情報が利用できない場合は、Google サービスアカウント認証情報を直接渡すことができます。すべての認証オプションについては、Pub/Subへのアクセスを構成するを参照してください。

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

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

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

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

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

データソースによっては、同等のSQLサポートが提供されていない場合があります。これらのデータ ソースのいずれかでレイクハウスフェデレーションを使用できない場合は、 Python使用してソースからデータを取り込むことができます。 PythonとSQLのソースファイルを同じパイプラインに追加できます。次の例では、リモートのPostgreSQLテーブル内のデータの現在の状態にアクセスするためのマテリアライズドビューを宣言します。

Python
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のロード構文を使用して、小さなまたは静的なデータセットをロードできます。パイプラインは、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"))
注記

read_files SQL 関数は、Databricks 上のすべての SQL 環境に共通です。これは、パイプラインで SQL を使用して直接ファイルにアクセスする場合に推奨されるパターンです。詳細については、 「オプション」を参照してください。

Pythonカスタムデータソースからデータをロードする​

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

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

Python
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の「忘れられる権利（right to be forgotten）」処理など）、skipChangeCommitsフラグを設定することで、それらの変更を無視することができます。このフラグは、option()関数を使用するspark.readStreamでのみ機能し、ソース ストリーミングテーブルがcreate_auto_cdc_flow（）関数のターゲットである場合は使用できません。詳細については、「ソースDelta Lakeテーブルへの変更の処理」を参照してください。

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

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

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

次の例では、シークレットを使用して、 Auto Loaderを使用してAzureデータレイク ストレージ アカウントから入力データを読み取るために必要なアクセス キーを保存します。 この同じ方法を使用して、たとえばS3にアクセスするためのAWSキーやApache Hive metastoreのパスワードなど、必要なシークレットを設定できます。

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

注記

シークレット値を設定するspark_conf構成キーにspark.hadoop.プレフィックスを追加する必要があります。

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

このコードサンプルでは、以下の値を置き換えてください。

プレースホルダー

に置き換える

<container-name>

Azureストレージアカウントコンテナーの名前。

<storage-account-name>

ADLSストレージアカウント名。

<path>

パイプラインの出力データとメタデータのパス。

<scope-name>

Databricksのシークレットスコープ名。

<secret-name>

Azureストレージアカウントのアクセスキーを含むキーの名前。

プレースホルダー

に置き換える

<container-name>

Azureストレージアカウントコンテナーの名前。

<storage-account-name>

ADLSストレージアカウント名。

<path>

パイプラインの出力データとメタデータのパス。

<scope-name>

Databricksのシークレットスコープ名。

<secret-name>

Azureストレージアカウントのアクセスキーを含むキーの名前。

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

このコードサンプルでは、以下の値を置き換えてください。

プレースホルダー

に置き換える

<container-name>

入力データを保存するAzureストレージアカウントコンテナーの名前。

<storage-account-name>

ADLSストレージアカウント名。

<path-to-input-dataset>

入力データセットへのパス。

プレースホルダー

に置き換える

<container-name>

入力データを保存するAzureストレージアカウントコンテナーの名前。

<storage-account-name>

ADLSストレージアカウント名。

<path-to-input-dataset>

入力データセットへのパス。

このページの見出し