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

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

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

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

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

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

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("s3://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権限を付与する必要があります。

Amazon S3からデータを取り込むには、S3バケットを参照するストレージ認証情報によって裏付けられた外部ロケーションを設定します。詳細については、 Unity Catalogを使用してクラウド オブジェクト ストレージに接続する」を参照してください。

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

メッセージバスからデータを取り込むようにパイプラインを構成できます。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():
  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()
  )

Databricks は、認証オプションを提供する際にシークレットを使用することを推奨します。すべての認証オプションについては、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>

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

このページの見出し