Databricks SQL Statement Execution API – Announcing the Public Preview - The Databricks Blogの翻訳です。
本書は抄訳であり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
本日、AWSとAzureで利用できるDatabricks SQL Statement Execution API(SQL文実行API)のパブリックプレビューを発表できることを嬉しく思っています。Databricksレイクハウスプラットフォームで管理されているでエータにアクセス、操作するためにREST API経由でDatabricks SQLのウェアハウスに接続することができるようになります。
Databricks SQL Statement Execution APIは、データへのアクセスをシンプルにし、皆様の要件に合わせたデータアプリケーションの構築を容易にします。このAPIは非同期型であり、JDBCやODBCの際に必要な接続の管理は不要です。さらに、ドライバーをインストールすることなしにSQLウェアハウスに接続できます。従来型のアプリケーションやクラウドベースのアプリケーション、サービス、デバイスをDatabricks SQLに接続するためにStatement Execution APIを活用することができます。また、お好きなプログラミング言語でカスタムクライアントライブラリを作成することができます。
この記事では、パブリックプレビューで利用できる主要機能のいくつかをレビューし、Statement Execution APIとJavaScriptを用いて、スプレッドシートのデータをどのように活用するのかをお見せします。
Statement Execution APIの概要
Statement Execution APIは、以下のオペレーションのHTTPエンドポイント経由で、SQL文の実行とDatabricks SQLのウェアハウスのすべてのタイプの処理結果データの取得を管理します。
| 実行するSQL文の送信 | POST /sql/statements |
| ステータスのチェック、結果の取得 | GET /sql/statements/{statement_id} |
| SQL文実行のキャンセル | POST /sql/statements/{statement_id}/cancel |
例えば、お使いのデータアプリケーションで表示するために、今年における月間注文の利益を取得したいとします。注文データはすでにレイクハウスで管理されているとすると、SQL文は以下のようなものになるでしょう。
SELECT
month(o_orderdate) as Month,
sum(o_totalprice) as `Monthly Revenue`
FROM `samples`.`tpch`.`orders`
WHERE year(o_orderdate) == year(current_date())
GROUP BY 1
ORDER BY 1
/api/2.0/sql/statementsエンドポイントにリクエストをPOSTすることで、SQL文の実行を起動することができます。SQL文の文字列表現は、SQLを実行するために使用されるSQLウェアハウスのIDとともにリクエストボディのペイロードとして指定されます。また、HTTPリクエストにはお使いのDatabricksワークスペースURLのホスト部分と認証のためのアクセストークンを含める必要があります。
POST /api/2.0/sql/statements HTTP/1.1
Host: <your_HOST>
Authorization: Bearer <your_access_token>
Content-Type: application/json
{
"statement": "SELECT month(o_orderdate) as Month, sum(o_totalprice) as `Monthly Revenue` FROM `samples`.`tpch`.`orders` WHERE year(o_orderdate) == year(current_date()) GROUP BY 1 ORDER BY 1",
"warehouse_id": "<your_SQL_warehouse_ID>"
}
SQL文がクイックに完了する場合、APIはPOSTリクエストに対する直接のレスポンスとして結果を返却します。以下にレスポンスのサンプルを示します。
{
"statement_id": "01ed9a57-ad5e-1333-8e76-8c718723abf2",
"status": {
"state": "SUCCEEDED"
},
"manifest": {
"format": "JSON_ARRAY",
"schema": {
"column_count": 2,
"columns": [
{
"name": "Month",
"type_name": "INT",
"position": 0
},
{
"name": "Monthly Revenue",
"type_name": "DECIMAL",
"position": 1,
"type_precision": 28,
"type_scale": 2
}
]
}
},
"result": {
"chunk_index": 0,
"row_offset": 0,
"row_count": 2,
"data_array": [
[
"1",
"14615808096.95"
],
[
"2",
"945871268.15"
]
]
}
}
SQL文の処理に時間を要する場合、APIは非同期で処理を続行します。この場合、レスポンスにはステートメントのIDとステータスが含まれます。
{
"statement_id": "01ed9a50-c9c9-178e-9be7-0ab52bc998b0",
"status": {
"state": "PENDING"
}
}
実行状態をチェックするためにステートメントIDを活用することができ、準備ができた際には、/api/2.0/sql/statements/{statement_id}エンドポイントにGETリクエストを送信することで結果を取得することができます。
GET /api/2.0/sql/statements/<statement_id> HTTP/1.1
Host: <your_HOST>
Authorization: Bearer <your_access_token>
また、/api/2.0/sql/statements/cancelエンドポイントにPOSTリクエストを送信して、リクエストをキャンセルするためにステートメントIDを使用することができます。
POST /api/2.0/sql/statements/<statement_id>/cancel HTTP/1.1
Host: <your_HOST>
Authorization: Bearer <your_access_token>
このAPIでは、リクエストをさらに設定することで同期的、非同期的に動作させることができます。詳細はチュートリアル(AWS | Azure)とドキュメント(AWS | Azure)を参照ください。
JavaScriptでDatabricks SQL Statement Execution APIを使う
任意のプログラミング言語からDatabricks SQL Statement Execution APIのリクエストを送信することができます。JavaScriptのFetch APIやPythonのPython Requests、Goのnet/httpパッケージなどのような手段を活用できます。
Spreadsheet AppからJavaScriptのFetch APIを用いてGoogleシートにデータを書き込むために、どのようにStatement Execution APIを活用できるのかをご説明します。
サンプルのスプレッドシートアプリケーション
注文データがスプレッドシートに入力されたGoogle Spreadsheetのアプリケーションを構築したいものとします。我々のユーザーは、今月や今年、ある期間における月間注文収益のように事前に定義された評価指標に基づいて月間収益データを取得することができます。それぞれの評価指標に対して、対応するSQL文を記述し、Statement Execution APIを用いて実行のための送信を行い、結果を取得、操作することができます。
次のセクションでは、このサンプルを実装するための主要なビルディングブロックの概要を説明します。GitHubからスプレッドシートをダウンロードすることができます。
スプレッドシートアプリケーションの構築
SQL Statement Execution APIを用いて実行したいSQL文を用いて、以下のexecuteStatement関数はAPIのデフォルトモードを取り扱う全体のロジックを含んでいます。このモードでは、SQL文の実行は同期的に処理をスタートし、デフォルトの10秒のタイムアウトの後、非同期的に処理を続行します。
submitStatement関数を用いて実行するSQL文を送信するところからスタートします。定義されたタイムアウト時間内に文の処理が完了した場合には、handleResult関数を呼び出すことで結果を取得します。そうでない場合には、実行処理は非同期的に続行され、完了するまで実行ステータスをポーリングする必要があることを意味します。checkStatus関数が必要なロジックをカバーします。終了すると、同じhandleResult関数を用いて結果を取得します。
function executeStatement(statement) {
var response = submitStatement(statement);
if (response.status.state == "SUCCEEDED") {
return handleResult(response.manifest, response.result);
} else {
response = checkStatus(response.statement_id)
while (response.status.state == "PENDING" || response.status.state == "RUNNING"
) {
response = checkStatus(response.statement_id)
}
return handleResult(response.manifest, response.result);
}
}
submitStatement関数は、10秒の待ち時間のタイムアウト(デフォルト)のような実行時のパラメーター、実行モード、SQL文を含むリクエストボディを定義します。さらに、HTTPリクエストを送信するために汎用的なfetchFromUrl関数を呼び出します。また、Databricksワークスペースユーザーのアクセストークンを引き渡すためにHTTP_REQUEST_BASE定数を定義します。送信するHTTPリクエストすべてでこの定数を再利用します。
const HTTP_REQUEST_BASE = {
headers: {
Authorization: `Bearer ${AUTH_TOKEN}`,
},
contentType: "application/json",
method: "GET",
payload: null,
muteHttpExceptions: true,
};
function submitStatement(statement) {
let body = {
"statement": statement,
"warehouse_id": WAREHOUSE,
"wait_timeout": "10s",
"on_wait_timeout": "CONTINUE",
};
let request = Object.assign({}, HTTP_REQUEST_BASE, { method: "POST", payload: JSON.stringify(body) });
let response = fetchFromUrl(`https://${HOST}/api/2.0/sql/statements`, request);
if (response.status.state == "FAILED") {
showError(`Submit request failed with unexpected state: ${response.status.state}`)
return null;
}
return response;
}
fetchFromUrl関数は、以下のように最低限のエラーハンドリングを持つHTTPリクエスト送信のための汎用関数です。
function fetchFromUrl(url, request) {
try {
let response = UrlFetchApp.fetch(url, request);
let responseJson = JSON.parse(response);
let statusCode = response.getResponseCode();
switch (statusCode) {
case 200:
return responseJson;
default:
showError(`Error: code=${responseJson["error_code"]} message=${responseJson["message"]}`);
return null;
}
} catch (error) {
showError(`Error: error=${error}`);
return null;
}
}
checkStatus関数では、待ち時間のタイムアウトを超えた場合、SQL文の実行ステータスを取得し、結果取得の準備が整ったのかどうかを特定するためにサーバーにポーリングを行います。
function checkStatus(statement_id) {
let response = fetchFromUrl(`https://${HOST}/api/2.0/sql/statements/${statement_id}`, HTTP_REQUEST_BASE);
if (response.status.state == "FAILED") {
let error = responseJson["status"]["error"]
showError(`Fetch request failed: code=${error["error_code"]} message=${error["message"]}`);
return null;
}
return response;
}
handleResult関数では、SQL文の処理が完了して結果の準備ができている場合には、取得のレスポンスには常に行の最初のチャンクが含まれます。この関数は結果をハンドリングし、以降のチャンクが利用できる場合には取得しようと試みます。
function handleResult(manifest, result) {
var columnNames = manifest["schema"]["columns"].map(col => col["name"]);
var chunks = [result.data_array];
while (result["next_chunk_internal_link"]) {
chunk = result["next_chunk_internal_link"];
result = fetchFromUrl(`https://${HOST}${chunk}`, HTTP_REQUEST_BASE);
chunks.push(result["data_array"]);
}
return [[columnNames]].concat(chunks).flat()
}
他に必要なことは、executeStatement関数を様々なユーザーインタフェースのウィジェットのJavaScriptイベントハンドラーに接続し、ユーザーの選択に対応するSQL文を引き渡すことです。GoogleのApps Scriptのドキュメントでは、返却データでスプレッドシートを埋めるための手順が提供されています。
Databricks SQL Statement Execution APIを使い始める
Databricks SQL Statement Execution APIは、Databricksのプレミアム、エンタープライズプランで利用できます。すでにDatabricksアカウントをお持ちである場合には、チュートリアル(AWS | Azure)とドキュメント(AWS | Azure)、コードサンプルのリポジトリをチェックしてください。まだ、Databricksをお使いでないのでしたらフリートライアルにサインアップしてください。
Databricks SQL Statement Execution APIは、お使いのDatabricks SQLウェアハウスに接続するための広範な選択肢を提供します。Python、Go、Node.js、CLI、ODBC/JDBCへのネイティブ接続の詳細に関しては、以前のブログ記事もチェックしてみてください。Databricksレイクハウスプラットフォームによって管理されるデータは本当にどこからでもアクセスできます！
さらにDatabricks SQL Statement Execution APIを学び、次にやってくるアップデートを知るには、Data + AI Summit 2023にご参加ください。