翻訳：Saki Kitaoka. - Original Blog Link
本日、AWS および Azure 上で Databricks SQL ステートメント実行 API の一般提供を開始することを発表します。このAPIを利用することで、REST API経由でDatabricks SQLウェアハウスに接続し、Databricks Lakehouse Platformが管理するデータにアクセスして操作することができます。
このブログでは、API の基本を説明し、GA リリースで新たに利用可能になる主要機能について説明し、 Databricks Python SDK を使用してステートメント実行 API を使用したデータアプリケーションを構築する方法を紹介します。Databricksワークスペースに対してコードを実行することで、そのコードを追跡することもできます。その他の例として、前回のブログでは、Statement Execution API と JavaScript を使用してスプレッドシートでデータを活用する方法を紹介しました。
BI アプリケーションはデータウェアハウスの主要なユースケースの一つであり、Databricks SQL はドライバ、コネクタ、および既存の BI ツールとのネイティブ統合による豊富な接続エコシステムを提供します。とはいえ、Databricks Lakehouse Platform が管理するデータは、eコマースプラットフォーム、CRM システム、SaaS アプリケーション、または顧客が社内で開発したカスタムデータアプリケーションなど、BI 以外のアプリケーションやユースケースにも関連しています。多くの場合、これらのツールは標準的なデータ ベースインターフェイスやドライバを介して簡単に接続することはできませんが、ほとんどすべてのツールやシステムはREST APIで通信することができます。
Databricks SQLステートメント実行APIを使用すると、HTTP上で標準SQLを使用して、幅広いアプリケーション、テクノロジー、およびコンピューティングデバイスとの統合を構築することができます。この API は、SQL 文を SQL Warehouse に送信して実行し、結果を取得するためのエンドポイント群を提供します。下図は典型的なデータフローの概要です。
APIを使用して、お好みのツールや言語でカスタムデータアプリケーションを構築できます。例えば、ビジネス・ユーザーがユーザー・インターフェースを通じてクエリ条件を提供し、その結果を可視化、ダウンロード、またはさらに分析するためのウェブ・アプリケーションを構築できます。また、APIを使用して、特定のユースケースやマイクロサービスに適合する特別な目的のAPIを実装したり、選択したプログラミング言語でカスタムコネクタを構築したりすることもできます。これらのシナリオでAPIを使用する場合の大きな利点の1つは、ドライバをインストールしたりデータベース接続を管理したりする必要がないことです。
AWSとAzureでAPIが一般的に利用できるようになるのに伴い、いくつかの新機能と改善が可能になります。
次のセクションでは、ステートメント実行APIの上にカスタムAPIを構築するために、これらの新機能を使用してさらに詳しく説明します。
専用APIの構築
今年の Data+AI Summitでは、この新しいステートメント実行APIを使ってDatabricks Lakehouse Platform上でカスタムAPIを構築する手順を紹介しました。今回は、Acme, Inc.という架空の会社のために、シンプルなウェブサイトとサービスのバックエンドを開発します。最初のステップとして `setup.sh`スクリプトを実行します。
Acme, Inc.は様々な種類の機械部品を販売する100の店舗を持つ中規模企業です。Databricks Lakehouse を活用して各店舗の情報を追跡し、メダリオンアーキテクチャで販売データを処理しています。店舗管理者がゴールドの販売データや店舗情報を簡単に閲覧できるウェブアプリケーションを作成したいと考えています。さらに、店舗管理者が通常のPOSを経由しなかった売上を挿入できるようにしたいと考えています。このシステムを構築するために、カスタムデータAPIを公開するPython Flaskアプリケーションと、そのAPIを呼び出してデータを読み書きするHTML/JQueryフロントエンドを作成します。
全店舗をリストアップするカスタムAPIエンドポイントと、それがバックエンドのステートメント実行APIにどのようにマッピングされるかを見てみましょう。これは引数を取らない単純なGETリクエストです。バックエンドは SQL Warehouse に静的な SELECT 文を呼び出して `stores` テーブルを読み取ります。
| Acme Inc's API Request | Statement Execution API Request | |
|---|---|---|
| GET /stores | → | POST /sql/statements statement: "SELECT * FROM stores" wait_timeout: "50s" on_wait_timeout: "CANCEL" |
| Acme Inc's API Response | Statement Execution API Response | |
| state: "SUCCEEDED" stores: [ ["123", "Acme, Inc", …], ["456", "Databricks", …] ] | ← | statement_id: "ID123" status: { state: "SUCCEEDED" } manifest: { ... } result: { data_array: [ ["123", "Acme, Inc", …], ["456", "Databricks", …] ] } |
Acmeには100店舗しかないため、高速なクエリと小さなデータセットのレスポンスを期待しています。そのため、Databricksに同期リクエストを行い、店舗データの行をインラインで返すことにしました。同期リクエストにするために、`wait_timeout` パラメータを設定し、レスポンスが返ってくるまで最大50秒待つことを指定し、それ以上かかる場合はクエリをキャンセルするように `on_wait_timeout` パラメータを設定しました。Databricks からのレスポンスを見ると、デフォルトの結果 `disposition` と `format` が JSON 配列にインラインでデータを返すことがわかります。Acmeのバックエンドサービスは、そのペイロードをリパッケージしてカスタムAPIの呼び出し元に返すことができます。
このカスタムエンドポイントの完全なバックエンドコードはここにあります。フロントエンドでは、カスタムエンドポイント `/api/1.0/stores` を呼び出してストアのリストを取得し、ここでJSON配列を反復処理することでストアを表示します。この2つで、AcmeはDatabricks SQLに支えられた新しいホームページを持つことになります！
また、各店舗について、直近の売上を表示するページを用意し、店舗管理者が店舗データ一式をダウンロードできるようにしたいと考えています。店舗ごとの売上トランザクションの数は、店舗数よりも何桁も多くなる可能性があります。このカスタムAPIエンドポイントの要件は以下の通りです：
以下に、Databricks SDK for Pythonを使用してステートメント実行APIを呼び出す方法を示します。全コードはこちらです。
def execute_list_sales_request(store_id, format, row_limit):
# Use parameters to prevent SQL injection via the store ID string.
parameters = [
StatementParameterListItem(name='store_id', value=store_id, type="INT")
]
statement_response = w.statement_execution.execute_statement(
statement = "SELECT * FROM sales WHERE store_id = :store_id",
format = Format[format],
disposition = Disposition.EXTERNAL_LINKS,
wait_timeout = "0s",
warehouse_id = warehouse_id,
parameters = parameters,
row_limit = row_limit
)
response = {
'request_id': statement_response.statement_id,
'state': str(statement_response.status.state.name)
}
return response最初の2つの要件を満たすために、カスタムAPIから`row_limit` と `format` パラメータを公開し、ステートメント実行APIに渡します。これにより、呼び出し元はクエリが生成する行の総数を制限したり、結果の形式（CSV、JSON、Arrow）を選択したりできるようになります。
カスタムAPIを非同期にするために、ステートメント実行APIの `wait_timeout` パラメータを0秒に設定します。つまり、Databricks はステートメント ID とクエリ状態を即座に応答します。このステートメント ID を `request_id`として、ステートメント状態と共に呼び出し元への応答にパッケージします。クライアントがリクエスト ID と状態を取得すると、同じカスタム API エンドポイントをポーリングして実行状況を確認することができます。エンドポイントは `get_statement` メソッド経由で Databricks SQL Warehouse にリクエストを転送します。クエリが成功すると、API は最終的に `SUCCEEDED` 状態と `chunk_count` を返します。チャンクカウントは、結果がいくつのパーティションに分割されたかを示します。
効率的な抽出（4番目の要件）を達成するために、EXTERNAL_LINKS dispositionを使用しました。これにより、各チャンクに対して事前に署名されたURLを取得することができ、`request_id` と `chunk_index`.が与えられるとカスタムAPIがそれを返します。
これを使用して、Acme に各店舗のランディングページを作成し、最新の売上を表示することができます。しかし、ページの上部に「ダウンロード」ボタンを追加して、ストアマネージャーがすべての売上の履歴データを引き出せるようにすることもできます。その場合、制限を設けず、CSV形式を活用することで、好きな分析ツールに簡単に取り込むことができます。ブラウザはクエリが成功し、チャンクの総数を取得すると、カスタムAPIを並行して呼び出し、事前に署名されたURLを取得し、クラウドストレージからCSVデータを直接ダウンロードします。EXTERNAL_LINKS処分は、逐次インライン読み取りよりも抽出スループットが12倍向上したCloud Fetchテクノロジーを活用しています。下の例では、500MBを160Mbpsで並行してダウンロードしました。
店舗の売上を表示できるようになったので、Acmeチームは新しい売上情報をLakehouseに挿入できるようにする必要があります。そのために、 /api/1.0/stores/storeId/sales endpoint へのPOSTリクエストでバックアップされたシンプルなWebフォームを作成します。フォームのデータをLakehouseに取り込むために、パラメータ化されたSQL文を使用します。
INSERT INTO
sales (
ss_sold_date_sk,
ss_ticket_number,
ss_store_sk,
ss_item_sk,
ss_sales_price,
ss_quantity
)
VALUES
(
:sold_date,
:sale_id,
:store_id,
:item_id,
:sales_price,
:quantity
)And supply the input from the web form to the Statement Execution API using the `parameters` list argument with a name, value, and type for each parameter:
{
"parameters": [
{ "name": "sold_date", "type": "DATE", "value": "2023-09-06" },
{ "name": "sale_id", "type": "BIGINT", "value": "10293847" },
{ "name": "store_id", "type": "INT", "value": "12345" },
{ "name": "item_id", "type": "INT", "value": "67890" },
{ "name": "sales_price", "type": "DECIMAL(7,2)", "value": "1.99" }
{ "name": "quantity", "type": "INT", "value": "100" },
]
}SQL WarehouseエンジンはSQLコードを解析した後、提供されたパラメータをリテラルとしてクエリプランに安全に代入します。これにより、悪意を持って注入されたSQL構文がSQLとして解釈されることを防ぎます。各パラメータの "type "フィールドは、提供された "value "の型の正しさをチェックすることで、さらなる安全性を提供します。悪意のあるユーザが数量フィールドの入力として "100); drop table sales"のような値を提供した場合、INSERT INTO文は以下のエラーとなり、実行されません。
[INVALID_PARAMETER_MARKER_VALUE.INVALID_VALUE_FOR_DATA_TYPE] An invalid parameter mapping was provided: the value '100); drop table sales' for parameter 'quantity' cannot be cast to INT because it is malformed.
ここでは、 `POST /api/1.0/stores/store_id/sales`エンドポイントの一部として使用するパラメータをどのように配置したかを見ることができます。ウェブフォームへの入力が正しいタイプで有効であれば、ユーザーが "Submit" をクリックした後、売上テーブルが正常に更新されます。
このカスタムAPIを繰り返し使用したり、Databricks Lakehouse Platform上で独自のカスタムデータアプリケーションを構築する足がかりとして使用することができます。この記事で使用したサンプルコードや、サンプルテーブルを作成する`setup.sh` スクリプトをご自身の Databricks 環境で使用するだけでなく、Data+AI Summit でのライブ説明 - 以下のビデオ - もご覧ください。
Databricks SQL 文実行 API は、Databricks Premium および Enterprise Tier で利用可能です。既に Databricks アカウントをお持ちの場合は、チュートリアル (AWS | Azure)、ドキュメント (AWS | Azure)、またはコードサンプルのリポジトリ(r repository)をご覧ください。まだ Databricks をご利用でない場合は、無料トライアル(free trial)にサインアップしてください。
