パイプラインで API からデータを取り込む

Ingesting from an API means pulling data over HTTP from a web service, usually as paginated JSON, rather than reading from a file or a database. Unlike files or a message bus, there's no built-in generic API source, so you handle authentication, pagination, and rate limits yourself. Lakeflow pipelines support three patterns for ingesting from an arbitrary API. Which one fits depends on your volume and refresh needs.

重要

カスタムAPI取り込みコードを作成する前に、ソース用のマネージドコネクタが既に存在するかどうかを確認してください。Lakeflow Connect には、Salesforce、Workday、ServiceNow、Google アナリティクスなど、一般的な多くのソフトウェア・アズ・ア・サービス（SaaS）API向けの組み込みコネクタが用意されており、パートナーコネクタのセットも拡充されています。コネクタがソースに対応している場合、認証、ページネーション、増分抽出が自動的に処理されるため、手動で取り込み処理を作成するよりも手間が大幅に省けます。Lakeflow Connectのマネージド コネクタを参照してください。以下のパターンは、適合するコネクタがない場合にのみ使用してください。

前提条件

• パイプライン。作成方法については、Lakeflow pipelinesのチュートリアルを参照してください。
• Databricks シークレットとして格納された、トークンやキーなどの API 資格情報。パイプラインのソースコードに認証情報をハードコーディングしないでください。「シークレット管理」を参照してください。
• Network access from your pipeline compute to the API endpoint.
• Familiarity with streaming tables and materialized views, the dataset types these patterns produce. See Streaming tables and Materialized views.

Choose a pattern

There's no native generic REST-API source in pipelines, so when you pull from an arbitrary API, pick one of three patterns based on data volume and how often you ingest:

パターン	Use when
周期的なプルをマテリアライズドビューとして	Payloads are small to medium and pulled once per pipeline run, such as reference data, daily FX rates, or a paginated but boundable API.
Python データソース API	You need to poll a high-volume or streaming API incrementally, with checkpointed progress so a restart doesn't re-read everything.
Auto Loader による分離された取り込み	You want to isolate API-specific quirks from your transformation logic and get exactly-once file tracking for free.

パターン	Use when
周期的なプルをマテリアライズドビューとして	Payloads are small to medium and pulled once per pipeline run, such as reference data, daily FX rates, or a paginated but boundable API.
Python データソース API	You need to poll a high-volume or streaming API incrementally, with checkpointed progress so a restart doesn't re-read everything.
Auto Loader による分離された取り込み	You want to isolate API-specific quirks from your transformation logic and get exactly-once file tracking for free.

Pattern 1: Periodic pulls as a materialized view

For small-to-medium payloads pulled once per pipeline run, write a Python function that calls the API and returns a Spark DataFrame. Because the dataset is a materialized view, the pipeline re-runs the function fully and idempotently every time the pipeline updates.

The following steps show you how to build a materialized view with periodic pulls:

1. Store the API token in a secret, then map it to a Spark configuration property in your pipeline settings so the pipeline code can read it. Add the property to the spark_conf block of the pipeline's cluster configuration:

   JSON

   {
     "clusters": [
       {
         "spark_conf": {
           "api.token": "{{secrets/<scope-name>/<secret-name>}}"
         }
       }
     ]
   }

   The code in the next step reads this value with spark.conf.get("api.token"). For more about configuring secrets in pipeline settings, see Securely access storage credentials with secrets in a pipeline.

2. API を呼び出し、その応答を DataFrame として返すマテリアライズドビューを定義します：

   Python

   import requests
   from pyspark import pipelines as dp
   from pyspark.sql import Row
   
   @dp.materialized_view(
       name="exchange_rates_bronze",
       comment="Daily FX rates pulled from a public REST API",
   )
   def exchange_rates_bronze():
       resp = requests.get(
           "https://api.example.com/v1/rates",
           params={"base": "USD"},
           headers={"Authorization": f"Bearer {spark.conf.get('api.token')}"},
           timeout=30,
       )
       resp.raise_for_status()
       rates = resp.json()["rates"]
       rows = [Row(currency=k, rate=float(v), as_of_date=resp.json()["date"]) for k, v in rates.items()]
       return spark.createDataFrame(rows)

3. ページをループさせて結果を連結してからDataFrameを返すために関数内でページを処理します:

   Python

   import requests
   from pyspark import pipelines as dp
   from pyspark.sql import Row
   
   @dp.materialized_view(
       name="customers_bronze",
       comment="Customers pulled from a paginated REST API",
   )
   def customers_bronze():
       token = spark.conf.get("api.token")
       rows = []
       url = "https://api.example.com/v1/customers"
       while url:  # follow the API's next-page cursor until exhausted
           resp = requests.get(
               url,
               headers={"Authorization": f"Bearer {token}"},
               timeout=30,
           )
           resp.raise_for_status()
           payload = resp.json()
           rows.extend(Row(**record) for record in payload["data"])
           url = payload.get("next")  # None on the last page
       return spark.createDataFrame(rows)

   Add retry and backoff logic around the request for resilience.

このパターンはパイプラインの更新ごとに API 応答全体を再読み込みするため、ペイロードが制限されている場合にのみ使用してください。増分読み取りには、パターン 2 を使用してください。

パターン 2: Python データソース API を使用した高ボリュームまたはストリーミング APIs

オフセット追跡を使用してインクリメンタルにポーリングする必要がある APIs については、Spark の Python Data Source API を使用してカスタムデータソースを実装してください。これにより、チェックポイントされた進捗状況やインクリメンタルな読み取りを含む適切なストリーミングセマンティクスが得られるため、再起動時に API 全体を再度プルするのではなく、最後のオフセットから再開されます。

The following steps show you how to ingest from a custom data source:

1. Implement a DataSource and DataSourceStreamReader that call the API and track the read offset. For details on authoring a custom data source, see PySpark custom data sources.

2. パイプラインがフォーマット名で参照できるように、データソースを登録します：

   Python

   spark.dataSource.register(MyApiDataSource)

3. Read from the registered source in a streaming table:

   Python

   from pyspark import pipelines as dp
   
   @dp.table(name="events_bronze")
   def events_bronze():
       return spark.readStream.format("my_api_source").load()

Pattern 3: Decouple ingestion with a scheduled job and Auto Loader

一般的な本番運用のパターンは、API 呼び出しをパイプラインから分離することです。スケジュールされたジョブは、生の API レスポンスをファイルとして Unity Catalog ボリュームに格納し、パイプラインが Auto Loader を使用してそれらを取得します。これにより、ページネーションやレート制限といった API 特有の癖を宣言型変換ロジックから分離し、Auto Loader の Exactly-Once（1 回のみ）ファイル追跡を無料で利用できるようになります。

次のステップでは、スケジュールされたジョブを使用してインジェストを分離する方法を説明します：

1. Write a notebook or script that calls the API and writes the raw JSON responses to a Unity Catalog volume. Read the API credentials from a secret. See Secret management.

   Python

   import requests, json, time
   
   token = dbutils.secrets.get(scope="<scope-name>", key="<secret-name>")
   volume_path = "/Volumes/main/raw/landing/api_events"
   
   resp = requests.get(
       "https://api.example.com/v1/events",
       headers={&quot;Authorization&quot;: f&quot;Bearer {token}&quot;},
       timeout=30,
   )
   resp.raise_for_status()
   # One file per run; the pipeline's Auto Loader tracks which files it has ingested.
   with open(f"{volume_path}/events_{int(time.time())}.json", "w") as f:
       json.dump(resp.json()["data"], f)

2. Lakeflow Jobsを使用して、ノートブックまたはスクリプトを自動的に実行するようにスケジュールします。See Lakeflow Jobs.

3. パイプライン内で、Auto Loader を使用して配置されたファイルを読み取るストリーミングテーブルを定義します。

   Python

   from pyspark import pipelines as dp
   
   @dp.table(name="api_events_bronze")
   def api_events_bronze():
       return (
           spark.readStream.format("cloudFiles")
               .option("cloudFiles.format", "json")
               .load("/Volumes/main/raw/landing/api_events")
       )

Auto Loader を使用した信頼性の高いファイル取り込みの詳細については、クラウド オブジェクト ストレージからのファイルの読み込みおよびAuto Loaderとはを参照してください。

Best practices for API ingestion

• シークレットをソースコードに含めないでください。 API トークンとキーを Databricks の Secret Scope に格納し、ランタイム時に読み取ります。「シークレット管理」を参照してください。
• Validate responses early. Add expectations on the ingested rows to catch malformed API responses before they flow downstream.
• Handle pagination and rate limits. Loop over pages and add retry with backoff so a transient failure doesn't fail the whole update.

その他のリソース

• Load data in pipelines
• Lakeflowコネクト のマネージド コネクタ
• PySpark custom data sources
• パイプラインのエクスペクテーションによるデータ品質の管理