パイプラインで API からデータを取り込む
API からの取り込みとは、ファイルやデータベースから読み取るのではなく、通常はページ分割された JSON として、Web サービスから HTTP 経由でデータを取得することを意味します。ファイルやメッセージバスとは異なり、組み込みの汎用 API ソースはないため、認証、ページネーション、レート制限は自身で処理する必要があります。LakeFlow Pipelines は、任意の API から取り込むための 3 つのパターンをサポートしています。どれが適しているかは、ボリュームと更新のニーズによって異なります。
カスタムAPI取り込みコードを作成する前に、ソース用のマネージドコネクタが既に存在するかどうかを確認してください。Lakeflow Connect には、Salesforce、Workday、ServiceNow、Google アナリティクスなど、一般的な多くのソフトウェア・アズ・ア・サービス（SaaS）API向けの組み込みコネクタが用意されており、パートナーコネクタのセットも拡充されています。コネクタがソースに対応している場合、認証、ページネーション、増分抽出が自動的に処理されるため、手動で取り込み処理を作成するよりも手間が大幅に省けます。Lakeflow Connectのマネージド コネクタを参照してください。以下のパターンは、適合するコネクタがない場合にのみ使用してください。
前提条件
- パイプライン。作成方法については、Lakeflow pipelinesのチュートリアルを参照してください。
- Databricks シークレットとして格納された、トークンやキーなどの API 資格情報。パイプラインのソースコードに認証情報をハードコーディングしないでください。「シークレット管理」を参照してください。
- パイプラインコンピュートからAPI Endpointへのネットワークアクセス。
- これらのパターンが生成するデータセット型である、ストリーミングテーブルとマテリアライズドビューに関する知識。ストリーミングテーブルおよびマテリアライズドビューを参照してください。
パターンを選択
パイプラインにはネイティブの汎用REST-APIソースは存在しないため、任意のAPIから取得する際は、データ量と取り込む頻度に基づいて3つのパターンから1つを選びます。
パターン | 使用する場合 |
|---|---|
ペイロードは小規模から中規模で、パイプラインのランごとに1回プルされます（参照データ、日次為替レート、ページネーションされるが境界設定可能なAPIなど）。 | |
再起動時にすべてを再読み込みしないように、チェックポイントで進捗を記録しながら、大容量またはストリーミング API を増分的にポーリングする必要があります。 | |
API固有の癖を変換ロジックから切り離し、ファイル追跡を一度だけ行う機能を無料で利用したいと考えているのですね。 |
パターン1:周期的なプルをマテリアライズドビューとして
パイプラインのランごとに1回プルされる小規模から中規模のペイロードについては、APIを呼び出してSpark DataFrameを返すPython関数を作成します。データセットはマテリアライズドビューであるため、パイプラインが更新されるたびに、パイプラインは関数を完全かつべき等に再実行します。
以下のステップで、定期的なプルを使ったマテリアライズドビューの構築方法を示します。
-
APIトークンをシークレットに保存し、パイプライン設定のSpark構成プロパティにマッピングすることで、パイプラインコードから読み取れるようにします。パイプラインのクラスター構成の
spark_confブロックにプロパティを追加します：JSON{
"clusters": [
{
"spark_conf": {
"api.token": "{{secrets/<scope-name>/<secret-name>}}"
}
}
]
}次のステップのコードは、
spark.conf.get("api.token")を使用してこの値を読み取ります。パイプライン設定でのシークレットの構成の詳細については、パイプラインでシークレットを使用してストレージ資格情報に安全にアクセスするを参照してください。 -
API を呼び出し、その応答を DataFrame として返すマテリアライズドビューを定義します：
Pythonimport 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) -
ページをループさせて結果を連結してからDataFrameを返すために関数内でページを処理します:
Pythonimport 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)回復力を高めるために、リクエストに再試行およびバックオフのロジックを追加します。
このパターンはパイプラインの更新ごとに API 応答全体を再読み込みするため、ペイロードが制限されている場合にのみ使用してください。増分読み取りには、パターン 2 を使用してください。
パターン 2: Python データソース API を使用した高ボリュームまたはストリーミング APIs
オフセット追跡を使用してインクリメンタルにポーリングする必要がある APIs については、Spark の Python Data Source API を使用してカスタムデータソースを実装してください。これにより、チェックポイントされた進捗状況やインクリメンタルな読み取りを含む適切なストリーミングセマンティクスが得られるため、再起動時に API 全体を再度プルするのではなく、最後のオフセットから再開されます。
次のステップでは、カスタムデータソースから取り込む方法を説明します：
-
APIを呼び出して読み取りオフセットを追跡する
DataSourceとDataSourceStreamReaderを実装します。カスタムデータソースの作成方法については、 PySparkカスタムデータソースを参照してください。 -
パイプラインがフォーマット名で参照できるように、データソースを登録します：
Pythonspark.dataSource.register(MyApiDataSource) -
ストリーミングテーブルで登録済みソースから読み取ります：
Pythonfrom pyspark import pipelines as dp
@dp.table(name="events_bronze")
def events_bronze():
return spark.readStream.format("my_api_source").load()
パターン3:スケジュールされたジョブと自動ローダーとの取り込みの分離
一般的な本番運用のパターンは、API 呼び出しをパイプラインから分離することです。スケジュールされたジョブは、生の API レスポンスをファイルとして Unity Catalog ボリュームに格納し、パイプラインが Auto Loader を使用してそれらを取得します。これにより、ページネーションやレート制限といった API 特有の癖を宣言型変換ロジックから分離し、Auto Loader の Exactly-Once（1 回のみ）ファイル追跡を無料で利用できるようになります。
次のステップでは、スケジュールされたジョブを使用してインジェストを分離する方法を説明します：
-
APIを呼び出し、Unityカタログのボリュームに生のJSON応答を書くノートブックやスクリプトを書きます。シークレットからAPIの認証情報を読み取る。「シークレット管理」を参照してください。
Pythonimport 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={"Authorization": f"Bearer {token}"},
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) -
Lakeflow Jobsを使用して、ノートブックまたはスクリプトを自動的に実行するようにスケジュールします。See Lakeflow Jobs.
-
パイプライン内で、Auto Loader を使用して配置されたファイルを読み取るストリーミングテーブルを定義します。
Pythonfrom 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とはを参照してください。
API 取り込みのベストプラクティス
- シークレットをソースコードに含めないでください。 API トークンとキーを Databricks の Secret Scope に格納し、ランタイム時に読み取ります。「シークレット管理」を参照してください。
- レスポンスを早期に検証します。 取り込まれた行に 期待値（expectations） を追加して、不正な形式の API レスポンスがダウンストリームに流れる前に捕捉します。
- ページ分割とレート制限を処理します。 ページをループ処理し、一時的な障害によって更新全体が失敗しないように、バックオフを伴う再試行を追加します。