オンラインからの利用を想定したDatabricksの性能検証

こんにちは、NRIデジタルの湯川です。
Azure Databricksはビッグデータ分析サービスですが、その活用のためのリファレンスアーキテクチャとして、Microsoftのドキュメントには以下の異なる２つのパターンの記載があります。

パターン１：分析レポート環境用のデータストアにCosmos DBなどDatabricks以外のサービスを利用する

Databricks によるストリーム処理 – Azure Reference Architectures

パターン２：分析レポート環境からDatabricksに直接SQLクエリを投げる（下記図の⑥＋⑦）

Azure Databricks を使用した最新の分析アーキテクチャ – Azure Architecture Center

オンラインシステムからクエリを受け付けるという場合、パターン１のように加工後、頻繁に利用するデータはCosmos DBやSQL Databaseのようなオンラインからの利用に強いDBに持たせるのではなく、パターン２のように「Databricksに直接SQLクエリで問い合わせを行った場合、どの程度の応答性能とスループットが出せるのか？」が今回のテーマです。

目次

• Azure Databricksとは
• 想定環境
• 性能検証の観点
• Databricks環境準備
• 検証データ
• 性能測定
  • 単純にレコードを１行持ってくる場合
    • 汎用クラスタ
    • SQL Warehouse
  • 単純にレコードを１行持ってくる場合の実行計画
  • 簡単な集計をさせた場合の実行計画
• 汎用クラスタとSQL Warehouseのコスト
  • 汎用クラスタの場合
  • SQL Warehouseの場合
• 性能測定結果評価
  • 汎用クラスタにおけるPhotonの有効性
  • 汎用クラスタとSQL Warehouseの差が想定より大きい
  • スケールアップ、スケールアウトの効果が非常に分かりやすい
  • Materialized Viewが待ち遠しい
• 結論

Azure Databricksとは

Azure Databricks はApache Spark ベースのビッグ データ分析サービスです。
Azureの場合、Sparkベースのビッグデータ分析はDatabricks以外にもAzure Synapse AnalyticsとAzure HDInsightがあります。Sparkの生みの親であるMatei Zaharia氏らが設立したDatabricks社とMicrosoftが、協業して提供しているAzureのファーストパーティサービスがAzure Databricksです。
Databricks社は数年遅れで自社で開発した新機能をOSS化しているものの、Photon^※など最新の機能を持ったSpark環境を使いたい場合はDatabricks社が提供しているAzure Databricks（以降、Databricksと記載した場合はサービスを指します）を選択することになります。

※PhotonはSparkを高速化するC++で開発されたベクトライズドエンジンです。PhotonについてはDatabricksの弥生さんが紹介記事を書かれていますのでこちらを参照ください。
Photonエンジン：Databricksレイクハウスプラットフォームにおける次世代クエリーエンジン – Qiit

想定環境

以下の図のように、AKSなどKubernetes上で動作するSpring Bootのアプリケーションから、JDBCでDatabricksに接続し、Delta Lake^※に保存されたデータに対してアクセスするケースを想定します。

※Delta LakeはParquetファイルにトランザクションログ（Delta Log）を追加することで、トランザクションやMVCCを実現したものです。

性能検証の観点

今回はオンラインシステムから普通のDBのように扱う場合を想定し、以下を前提としました。

• 応答時間を測定する。許容できる最大応答時間は5秒とする
• スループットを測定する。一部の限られたデータ分析担当者ではなく多数の人が利用する場合を想定

5秒以上の待ち時間を許容できるケースも、限られた人しか利用しないケースもあると思いますが、汎用的に利用できると言えるかどうかの判断基準として上記を条件としています。

Databricks環境準備

Azureの東日本リージョンにDatabricksのワークスペース^※1を用意し、汎用クラスタ^※2とSQL Warehouse^※3をポータルから作成します。
※1 ワークスペースはDatabricksの各種リソースを束ねる論理的な枠組みです。Databricksの各種リソースはワークスペースを経由して操作します。
※2 汎用クラスタは汎用的なSparkジョブを実行するためのクラスタです。
※3 SQL WarehouseもSparkジョブを実行するクラスタですが、SQLを使った操作に特化しています。

汎用クラスタは以下の画面で、PolicyをUnrestrictedに設定して作成しました。

SQL Warehouseは以下の画面から作成しています。

どちらもクラスタを作成すると、JDBCドライバで接続するためのエンドポイント情報が表示されます。

汎用クラスタの場合はAdvanced optionsにJDBC/ODBCというタブがあるので選択し、JDBC URLの2.6.25 or laterを選びます。以下のキャプチャでは消しています。

SQL Warehouseの場合はConntection detailsに表示されます。以下のキャプチャでは消しています。

接続文字列にはトークンが必要なので、User SettingsからGenerate new tokenを選択し、トークンを作成します。

JDBCドライバはbuild.gradleに以下のように記載し、Spring Boot＋Mybatisから利用するようにしました。
springboot-databricks-rest-api-master.zip

HikariCPのコネクション数は30にしています。

implementation 'com.databricks:databricks-jdbc:2.6.32'

Javaの起動時に以下のオプションがないとJDBCの初期化に失敗するので追加します。

--add-opens=java.base/java.nio=ALL-UNNAMED

検証データ

検証に使用するデータはSynapse Link for SQLの検証をした時のデータを流用することにしました。
Azure SQL Databaseに以下のテーブルを定義し、sys.objectsを増幅させてテストデータとしています。
件数は1000万件です。

CREATE TABLE [dbo].[test_data_10000000](
	[name] [sysname] NOT NULL,
	[object_id] [int] NOT NULL,
	[principal_id] [int] NULL,
	[schema_id] [int] NOT NULL,
	[parent_object_id] [int] NOT NULL,
	[type] [char](2) NULL,
	[type_desc] [nvarchar](60) NULL,
	[create_date] [datetime] NOT NULL,
	[modify_date] [datetime] NOT NULL,
	[is_ms_shipped] [bit] NOT NULL,
	[is_published] [bit] NOT NULL,
	[is_schema_published] [bit] NOT NULL,
	[id] [bigint] IDENTITY(1,1) NOT NULL,　
	pkey AS (id % 10) PERSISTED --計算列を定義する場合
 CONSTRAINT [pk_test_data] PRIMARY KEY CLUSTERED 
(
	[id] ASC
));

Azure SQL DatabaseのデータをDelta Tableに取り込みます。この作業はDatabricksノートブックを使って実行しました。
まずは外部テーブルとして、SQL Server上のテーブルを定義します。

CREATE TABLE sqldb
USING org.apache.spark.sql.jdbc
OPTIONS (
  url "jdbc:sqlserver://sql-synapsepoc.database.windows.net:1433;database=sqldb-dev-synapsepoc;selectMethod=cursor;",
  dbtable "[dbo].[test_data_10000000]",
  user "xxxxxxxxxxxxxxxxxxxxxxx",
  password "xxxxxxxxxxxxxxxxxxxx"
)

Delta Tableを定義します。

CREATE TABLE testdata (name	string,
  object_id	int,
  principal_id	int,
  schema_id	int,
  parent_object_id	int,
  type	string,
  type_desc	string,
  create_date	timestamp,
  modify_date	timestamp,
  is_ms_shipped	boolean,
  is_published	boolean,
  is_schema_published	boolean,
  id	bigint
);

Delta Tableに流し込みます。

MERGE INTO default.testdata 
USING sqldb
ON testdata.object_id = sqldb.object_id
WHEN MATCHED THEN
  UPDATE SET
    id = sqldb.id,
    name = sqldb.name,
    object_id = sqldb.object_id,
    principal_id = sqldb.principal_id,
    schema_id = sqldb.schema_id,
    parent_object_id = sqldb.parent_object_id,
    type = sqldb.type,
    type_desc = sqldb.type_desc,
    create_date = sqldb.create_date,
    modify_date = sqldb.modify_date,
    is_ms_shipped = sqldb.is_ms_shipped,
    is_published = sqldb.is_published,
    is_schema_published = sqldb.is_schema_published
WHEN NOT MATCHED
  THEN INSERT 
  (id, name, object_id, principal_id, schema_id, parent_object_id, type, type_desc, create_date, modify_date, is_ms_shipped, is_published, is_schema_published)
  VALUES (
  sqldb.id, 
  sqldb.name, 
  sqldb.object_id, 
  sqldb.principal_id, 
  sqldb.schema_id, 
  sqldb.parent_object_id, 
  sqldb.type, 
  sqldb.type_desc, 
  sqldb.create_date, 
  sqldb.modify_date, 
  sqldb.is_ms_shipped, 
  sqldb.is_published, 
  sqldb.is_schema_published
  )

以上でデータの準備ができました。

性能測定

単純にレコードを１行持ってくる場合

以下のような非常に単純なSQLで応答時間とスループットを確認してみます。
パラメータはJavaのコード内でランダムになるようにして、Apache Benchを使って負荷をかけました。

SELECT id, name, object_id, principal_id, schema_id, parent_object_id, type, type_desc, is_ms_shipped, is_published, is_schema_published FROM default.testdata WHERE id = #{id}

汎用クラスタ

Photonを有効にしていない場合は通常時でも応答に1秒程度かかるようです。
Photonを有効にすることで、応答時間は500-600ms程度に短縮されました。
D4sv5１台（Photon有効）の場合、秒間の処理件数は5.25程度で頭打ちになるようです。
D4sv5は4コア16GBメモリのVMですが、これをD8sv5（8コア32GBメモリ）のVMにスケールアップしても、台数を2倍に増やしても、秒間の処理件数はほぼ単純に倍になりました。

SQL Warehouse

SQL WarehouseにはPhotonの有効無効という選択肢がなく、クラスターサイズと台数のみを選択します。
今回、クラスターサイズは最小の2X-Smallのみを試しました。Azureのドキュメントによると、これはE8dsv4（8コア64GBメモリ）に相当するようです。

結果ですが、応答時間は汎用クラスタより速く、450msから550msに収まりました。
1台の場合、秒間の処理件数は22程度で頭打ちになるようです。
台数を2台にすると、秒間の処理件数は倍になりました。

単純にレコードを１行持ってくる場合の実行計画

SQL Warehouseで先ほどの単純にレコードを１行持ってくる場合の実行計画をみてみます。
対象のデータを探す部分より、Columnar To Rowの方に時間がかかっているようです。

簡単な集計をさせた場合の実行計画

次に以下のような簡単な集計をさせた場合を試してみます。

SELECT name, COUNT(id) AS cnt_id FROM testdata t GROUP BY name LIMIT 1000

2X-Small１台の場合ですが、結果は次のようになりました。
全体としては16秒程度かかっています。

汎用クラスタとSQL Warehouseのコスト

性能評価に入る前にコストの考え方について確認しておきたいと思います。

汎用クラスタの場合

汎用クラスタの場合、料金はVMと消費するDBUの合計になります。
D4sv5１台で消費するDBUは1。Photonを有効にすると2です。
以下が2022/12/30現在の料金表です。
Photonが有効なD4sv5２台のクラスタを１ヶ月Premiumレベルで稼働させると278,400円程度かかります。

SQL Warehouseの場合

SQL WarehouseではクラスタサイズとDBUで料金が決まります。
2X-Smallの場合、スケールが１でもDBUは4消費します。
以下が2022/12/30現在の料金表です。
2X-Smallを1ヶ月稼働させたままにすると、234,360円程度かかります。

性能測定結果評価

汎用クラスタにおけるPhotonの有効性

オンラインから汎用クラスタにクエリを投げる場合はPhotonを有効にした方が良いでしょう。
オンラインで１回の応答にかかる時間が500ms程度なのか1秒程度なのかは大きな違いだと思います。

汎用クラスタとSQL Warehouseの差が想定より大きい

今回、一番驚いたのは汎用クラスタとSQL Warehouseの違いです。おおよそ同じ費用の環境でSQL Warehouseの方が応答速度で50ms-100ms程度勝り、秒間で倍以上の量を処理することができました。
Databricks社のエンジニアさんからもSQLのみを実行するのであれば、SQL Warehouseを使ってほしいと言われていましたが、同じSparkベースの基盤でここまで差があったのは驚きでした。

スケールアップ、スケールアウトの効果が非常に分かりやすい

汎用クラスタでもSQL Warehouseでもスケールアップ、スケールアウトで非常に分かりやすくスループットが改善しました。
サイジングを考える上で、この単純さは非常にありがたいなと思います。

Materialized Viewが待ち遠しい

1000万件程度の簡単な集計結果を返すのに単発で16秒かかりました。
この時間が許容できない場合は事前に集計を終わらせておく必要があるでしょう。
DatabricksではMaterialized Viewがまだプレビューですが、非常に強力な武器になると思います。

結論

非常に簡易なシナリオではありますが、冒頭で掲げていたオンラインシステムからクエリを受け付けるという場合に「Databricksに直接SQLクエリで問い合わせを行った場合、どの程度の応答性能とスループットが出せるのか？」についてです。
事前に加工をすべて終えられる前提で単純に数行とってくるようなSELECT文を投げるようなシナリオの場合、SQL Warehouseであれば現実的な時間内に応答を返すことはできそうです。
ストレージの限界がどこかできてしまうのだろうと思いますが、スケールアップやスケールアウトでスループットを簡単に増強できるところも魅力です。
Materialized ViewがGAになってからまた試してみたいと思います。