Azure Data Lake Storage の Query Acceleration が GA になったので試したら最高だった

ブチザッキによると Build 2019 での Mark Russinovich 御大のセッションで発表されていた Blob の内容に対してクエリを実行できる、当時は Quick Query と呼ばれていた機能が Ignite 2020 前に GA していたようです。

今年の 5 月ぐらいから名前が Quick Query ではなく Query Acceleration に変わっていたようですが、長いし typo しそうなので Quick Query と言ってしまいそうです。

AWS の Athena に似た機能のように見えますが、Query Acceleration は事前にテーブルを作る必要がなく、クエリを投げるだけで使えるので簡単です。データフォーマットとしては CSV*1 と JSON を扱えます。

Ignite 2020 の "What’s New in Azure Storage" というセッションでも GA の発表がありました。Microsoft によると読み取られたデータの 20% 以下しか実際の分析クエリに使われないらしく、データに近い部分でフィルタリングを行えるようにすることで全体的な最適化を図る仕組みです。

セッション曰く "Deeply integrated into Azure Synapse Analytics" らしいのでかなり期待が持てます。Azure Storage 周りのアップデートは多すぎなのでセッションを見ておくと良いです。

機能名は Query Acceleration for Azure Data Lake Storage となっていますが、Hierarchical namespace が無効な GPv2 のストレージアカウントに対しても利用できるので、若干の分かりにくさを生んでいます。

SDK 的には Blob Storage と Data Lake Storage それぞれに名前は違いますが同じインターフェースで用意されているので、特に悩むことは無いでしょう。ぶっちゃけ Blob Storage の SDK で ADLS も使えます。

Query Acceleration の機能と価格

日本語版は GA 向けに更新されていないので、これまで通り基本は英語版を参照するようにすると間違いがありません。特に NuGet パッケージの扱い周りが全然違うので、日本語版を見ているとはまります。

ドキュメントは比較的充実しているので、一通り目を通しておきたいですね。特に SQL に関しては T-SQL とはある程度の互換性はありますが、当然ながら使えない機能が多いです。

フィルタリングを行うチュートリアルも用意されています。ドキュメントにも手順として記載されていますが、使うためには BlobQuery の Resource Provider を登録する必要があるので注意してください。

今後は自動で登録されると思いますが、少なくとも手持ちのサブスクリプションでは登録が必要でした。

新しい Azure SDK から 4 つの言語で使えるようになっています。PowerShell を使う例も載っていますが、Azure CLI でも az storage blob query コマンドが実装されているので、サクッと試す分には使えます。

そして気になる課金体系ですが、ドキュメントにはスキャンされたデータとクライアントに返されたデータに対して課金されるとあります。Pricing にも一応書かれていますが、単位が謎なので計算が出来ません。

こっちはまだプレビューの情報のままのようなので、更新されるのを待ちたいと思います。

Synapse Analytics の SQL on-demand や AWS の Athena を見ると課金体系としてはよくあるものなので、飛びぬけて高いということにはならないでしょう。

ここからは実際にアプリケーションから Query Acceleration を使ってみます。CSV と JSON で若干使い方が異なっているので、それぞれのフォーマットで試しています。

CSV に対してクエリを実行する

実際に Query Acceleration を C# SDK を使って試してみます。今回は Blob Storage SDK を使っていますが、前述の通り ADLS SDK では名前が変わっているだけなので適宜読み替えてください。

試すにあたってはそれなりのデータ量のファイルを用意しないと面白くないので探し回ったのですが、以下のサイトで国勢調査ベースの人口統計が CSV で手に入ったのでこれを使いました。

C# SDK は NuGet で公開されている 12.6.0 以上をインストールします。これが対応バージョンです。

最初はコンテナー単位でクエリが書けるのかと期待しましたが、Blob 単位でクエリを実行する必要があったので、大量のファイルからフィルタリングする場合には少し手間がかかります。

正直なところワイルドカードで Blob を指定できるとかなり良さそうでしたが、今回は 1 つのファイルに対してなので単純なコードで実現することにします。本質的な部分は BlobQueryOptions の用意と QueryAsync の実行だけです。それ以外はいつも通りのコードです。

var connectionString = "DefaultEndpointsProtocol=https;AccountName=***;AccountKey=***;EndpointSuffix=core.windows.net";

var blobServiceClient = new BlobServiceClient(connectionString);
var containerClient = blobServiceClient.GetBlobContainerClient("sampledata");

var blobClient = containerClient.GetBlockBlobClient("c03.csv");

var options = new BlobQueryOptions
{
    InputTextConfiguration = new BlobQueryCsvTextOptions
    {
        HasHeaders = false
    }
};

var result = await blobClient.QueryAsync("SELECT * FROM BlobStorage WHERE _7 < 1000", options);

var content = await new StreamReader(result.Value.Content).ReadToEndAsync();

Console.WriteLine(content);

InputTextConfiguration には BlobQueryCsvTextOptions を指定しています。このクラスはデリミタやエスケープシーケンスのプロパティを持っているので、必要に応じて変更しましょう。

本来なら HasHeaders を true に設定すると SQL からカラム名でアクセス出来るので便利ですが、後述する問題によって今回は false を設定しています。

カラムに対して名前でアクセスは出来ませんが _1, _2, _3 のようにインデックスでアクセス出来るので、これを使ってフィルタリングのクエリを書いています。今回の例では _7 は人口に該当します。

このコードを実行してみると、人口でフィルタリングされたデータが返ってくることが確認できます。

もちろん SELECT を使って必要なカラムに絞り込んだり、集計関数を 1 つだけ使うことも出来ます。CSV には型が無いので、集計関数を使うときはキャストが必要で少し面倒でした。

現時点で把握している問題としては、SQL 内でマルチバイト文字を使うとエラーになるという点です。従って CSV のヘッダーやフィルタリングする値が日本語の場合はエラーになります。

今回使用した国勢調査の人口統計はヘッダーや値に日本語が含まれているので、ヘッダーなしのデータとして扱うしか方法がありませんでした。フィードバック済みなので対応待ちです。

この程度の問題であれば Public Preview に気が付いていれば、GA する前に報告が出来たと思うので残念です。とは言えマルチバイト文字を使うケースは少ない思うのであまり実害はないでしょう。

JSON に対してクエリを実行する

次は JSON に対してのクエリを試します。対応するデータは 1 行が 1 つの JSON として表現されているものになるので、Azure Monitor が出力するログなどが分析対象としては面白いと考えるでしょう。

残念ながら Azure Monitor からのログは Append Blob が使われているので Query Acceleration は使えません。

もちろん何らかの方法で Block Blob としてコピーしなおせば使えますが、正直かなり面倒なので分析用は Log Analytics に入れて KQL を書いた方が良い気がします。

今回はデータを用意するのが面倒だったので App Service が Azure Monitor 経由で出力するアクセスログを、Block Blob としてコピーしなおしたものを使います。ファイルが 300 近くあるので Blob の一覧を取得して順次クエリを実行しています。

var connectionString = "DefaultEndpointsProtocol=https;AccountName=***;AccountKey=***;EndpointSuffix=core.windows.net";

var blobServiceClient = new BlobServiceClient(connectionString);
var containerClient = blobServiceClient.GetBlobContainerClient("httplogs");

await foreach (var blobItem in containerClient.GetBlobsAsync(BlobTraits.Metadata))
{
    var blobClient = containerClient.GetBlockBlobClient(blobItem.Name);

    var options = new BlobQueryOptions
    {
        InputTextConfiguration = new BlobQueryJsonTextOptions(),
        OutputTextConfiguration = new BlobQueryJsonTextOptions()
    };

    var result = await blobClient.QueryAsync("SELECT CsMethod, UserAgent FROM BlobStorage[*].properties WHERE CsUriStem = '/'", options);

    var content = await new StreamReader(result.Value.Content).ReadToEndAsync();

    Console.Write(content);
}

JSON を対象にする場合は InputTextConfiguration に明示的に BlobQueryJsonTextOptions を設定しないと、CSV として読み込もうとしてエラーになります。同様に OutputTextConfiguration にも指定しないと CSV として返ってきます。*2

クエリの説明はあまり要らないと思いますが、サイトのルート / へのログだけをフィルタリングして、HTTP Method と User-Agent だけ出力しています。

FROM の書き方があまり見ない形ですが、Azure Monitor からの JSON は以下のように properties でネストされた形になっているので、そのプロパティだけに絞り込むという指定をしています。

{
  "time": "2020-08-27T00:52:17.7598888Z",
  "resourceId": "/SUBSCRIPTIONS/00000000-0000-0000-0000-000000000000/RESOURCEGROUPS/DEFAULT-WEB-JAPANEAST/PROVIDERS/MICROSOFT.WEB/SITES/APPSERVICEINFO",
  "category": "AppServiceHTTPLogs",
  "properties": {
    "CsMethod": "GET",
    "CsUriStem": "/api/siteextension",
    "SPort": "443",
    "CIp": "0.0.0.0",
    "UserAgent": "Mozilla/5.0+(Windows+NT+10.0;+Win64;+x64)+AppleWebKit/537.36+(KHTML,+like+Gecko)+Chrome/86.0.4223.0+Safari/537.36+Edg/86.0.608.2",
    "CsHost": "appservice.info",
    "ScStatus": 200,
    "ScSubStatus": "0",
    "ScWin32Status": "0",
    "ScBytes": 1188,
    "CsBytes": 1205,
    "TimeTaken": 46,
    "Result": "Success",
    "Cookie": "-",
    "CsUriQuery": "X-ARR-LOG-ID=00000000-0000-0000-0000-000000000000",
    "CsUsername": "-",
    "Referer": "https://appservice.info/"
  }
}

このコードを実行してみると、順次該当するログが出力されていきます。

Blob 単位でのクエリになるので、良い感じのパス規約や Blob Index Tags と組み合わせると、必要なものだけ取れるようにするとさらに効率的でしょう。

最近は Synapse Analytics の SQL on-demand で Azure Data Lake Storage に保存した CSV に対してクエリを書きたいことが多かったのですが、複雑なクエリが必要なくアプリケーションから使いたい場合に Query Acceleration はかなり便利だと思います。

実際に仕事でも早速 Query Acceleration を使いたい場面があるので、更に利用できる SQL の検証や ADLS の設計に励んでいきたいところです。

*1:デリミタなどはクエリ時に指定できるので TSV なども可能のはず

*2:CSV => JSON / JSON=>CSV にも使えそうな感じがする