Azure Data Lake Storage クエリアクセラレーションを使用してデータをフィルター処理する

2025-04-09

この記事では、クエリアクセラレーションを使用して、ストレージアカウントからデータのサブセットを取得する方法について説明します。

クエリアクセラレーションを使用すると、アプリケーションと分析フレームワークは、特定の操作を実行するために必要なデータのみを取得することで、データ処理を大幅に最適化できます。詳細については、 Azure Data Lake Storage のクエリ高速化に関するページを参照してください。

前提条件

Azure Storage にアクセスするには、Azure サブスクリプションが必要です。まだサブスクリプションをお持ちでない場合は、開始する前に無料アカウントを作成してください。
汎用 v2 ストレージアカウント。ストレージアカウントの作成を参照してください。
二重暗号化はサポートされていません。
JSON ファイルに対してクエリを実行する場合、このファイル内の各レコードサイズは 1 MB 未満にする必要があります。
タブを選択すると、SDK 固有の前提条件が表示されます。
- PowerShell
- .NET
- Java
- Python
- Node.js
適用なし

.NET SDK
- Java Development Kit (JDK) バージョン 8 以降
- Apache Maven
  
  注
  
  この記事では、Apache Maven を使用して Java プロジェクトを作成したことを前提としています。 Apache Maven を使用してプロジェクトを作成する方法の例については、「設定」を参照してください。
Python 3.8 以降。

Node.js SDK を使用するために必要な追加の前提条件はありません。

環境を設定する

手順 1: パッケージをインストールする

Az モジュールバージョン 4.6.0 以降をインストールします。

Install-Module -Name Az -Repository PSGallery -Force

以前のバージョンの Az から更新するには、次のコマンドを実行します。

Update-Module -Name Az

コマンドプロンプトを開き、ディレクトリ (cd) をプロジェクトフォルダーに変更します。次に例を示します。
```
cd myProject
```
dotnet add package コマンドを使用して、.NET パッケージ用 Azure Blob Storage クライアントライブラリの 12.5.0-preview.6 バージョン以降をインストールします。
```
dotnet add package Azure.Storage.Blobs -v 12.8.0
```
この記事に示す例では、 CsvHelper ライブラリを使用して CSV ファイルを解析します。そのライブラリを使用するには、次のコマンドを使用します。
```
dotnet add package CsvHelper
```

プロジェクトの pom.xml ファイルをテキストエディターで開きます。依存関係のグループに、次の dependency 要素を追加します。

<!-- Request static dependencies from Maven -->
<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-core</artifactId>
    <version>1.6.0</version>
</dependency>
 <dependency>
     <groupId>org.apache.commons</groupId>
     <artifactId>commons-csv</artifactId>
     <version>1.8</version>
 </dependency>
 <dependency>
   <groupId>com.azure</groupId>
   <artifactId>azure-storage-blob</artifactId>
   <version>12.8.0-beta.1</version>
 </dependency>

pip を使用して、Python 用 Azure Data Lake Storage クライアントライブラリをインストールします。

pip install azure-storage-blob==12.4.0

ターミナルウィンドウを開いてから次のコマンドを入力して、JavaScript 用の Data Lake クライアントライブラリをインストールします。

    npm install @azure/storage-blob
    npm install @fast-csv/parse

手順 2: ステートメントを追加する

適用なし

これらの using ステートメントをコードファイルの先頭に追加します。

using Azure.Storage.Blobs;
using Azure.Storage.Blobs.Models;
using Azure.Storage.Blobs.Specialized;

クエリアクセラレーションは、CSV と JSON 形式のデータを取得します。そのため、使用する CSV または JSON 解析ライブラリの using ステートメントを必ず追加してください。この記事に示す例では、NuGet で使用できる CsvHelper ライブラリを使用して CSV ファイルを解析します。したがって、これらの using ステートメントをコードファイルの先頭に追加します。

using CsvHelper;
using CsvHelper.Configuration;

この記事で紹介する例をコンパイルするには、これらの using ステートメントも追加する必要があります。

using System.Threading.Tasks;
using System.IO;
using System.Globalization;

これらの import ステートメントをコードファイルの先頭に追加します。

import com.azure.storage.blob.*;
import com.azure.storage.blob.options.*;
import com.azure.storage.blob.models.*;
import com.azure.storage.common.*;
import java.io.*;
import java.util.function.Consumer;
import org.apache.commons.csv.*;

これらの import ステートメントをコードファイルの先頭に追加します。

import sys, csv
from azure.storage.blob import BlobServiceClient, ContainerClient, BlobClient, DelimitedTextDialect, BlobQueryError

コードファイルの先頭にこのステートメントを配置して、 storage-blob モジュールを含めます。

const { BlobServiceClient } = require("@azure/storage-blob");

クエリアクセラレーションは、CSV と JSON 形式のデータを取得します。そのため、使用する CSV または JSON 解析モジュールのステートメントを必ず追加してください。この記事に示す例では、 fast-csv モジュールを使用して CSV ファイルを解析します。したがって、このステートメントをコードファイルの先頭に追加します。

const csv = require('@fast-csv/parse');

フィルターを使用してデータを取得する

SQL を使用して、クエリ高速化要求で行フィルター述語と列プロジェクションを指定できます。次のコードは、ストレージ内の CSV ファイルに対してクエリを実行し、3 番目の列が Hemingway, Ernest値と一致するすべてのデータ行を返します。

SQL クエリでは、クエリ対象のファイルを示すためにキーワード BlobStorage が使用されます。
最初の列が_1である列参照は_Nとして指定されます。ソースファイルにヘッダー行が含まれている場合は、ヘッダー行で指定された名前で列を参照できます。

Function Get-QueryCsv($ctx, $container, $blob, $query, $hasheaders) {
    $tempfile = New-TemporaryFile
    $informat = New-AzStorageBlobQueryConfig -AsCsv -HasHeader:$hasheaders
    Get-AzStorageBlobQueryResult -Context $ctx -Container $container -Blob $blob -InputTextConfiguration $informat -OutputTextConfiguration (New-AzStorageBlobQueryConfig -AsCsv -HasHeader) -ResultFile $tempfile.FullName -QueryString $query -Force
    Get-Content $tempfile.FullName
}

$container = "data"
$blob = "csv/csv-general/seattle-library.csv"
Get-QueryCsv $ctx $container $blob "SELECT * FROM BlobStorage WHERE _3 = 'Hemingway, Ernest, 1899-1961'" $false

非同期メソッド BlockBlobClient.QueryAsync クエリアクセラレーション API にクエリを送信し、結果を Stream オブジェクトとしてアプリケーションにストリーミングします。

static async Task QueryHemingway(BlockBlobClient blob)
{
    string query = @"SELECT * FROM BlobStorage WHERE _3 = 'Hemingway, Ernest, 1899-1961'";
    await DumpQueryCsv(blob, query, false);
}

private static async Task DumpQueryCsv(BlockBlobClient blob, string query, bool headers)
{
    try
    {
        var options = new BlobQueryOptions()
        {
            InputTextConfiguration = new BlobQueryCsvTextOptions()
            { 
                HasHeaders = true, 
                RecordSeparator = "\n", 
                ColumnSeparator = ",", 
                EscapeCharacter = '\\', 
                QuotationCharacter = '"'
            },
            OutputTextConfiguration = new BlobQueryCsvTextOptions() 
            { 
                HasHeaders = true, 
                RecordSeparator = "\n", 
                ColumnSeparator = ",", 
                EscapeCharacter = '\\', 
                QuotationCharacter = '"' },
            ProgressHandler = new Progress<long>((finishedBytes) => 
                Console.Error.WriteLine($"Data read: {finishedBytes}"))
        };
        options.ErrorHandler += (BlobQueryError err) => {
            Console.ForegroundColor = ConsoleColor.Red;
            Console.Error.WriteLine($"Error: {err.Position}:{err.Name}:{err.Description}");
            Console.ResetColor();
        };
        // BlobDownloadInfo exposes a Stream that will make results available when received rather than blocking for the entire response.
        using (var reader = new StreamReader((await blob.QueryAsync(
                query,
                options)).Value.Content))
        {
            using (var parser = new CsvReader
                (reader, new CsvConfiguration(CultureInfo.CurrentCulture) { HasHeaderRecord = true }))
            {
                while (await parser.ReadAsync())
                {
                    Console.Out.WriteLine(String.Join(" ", parser.Parser.Record));
                }
            }
        }
    }
    catch (Exception ex)
    {
        System.Windows.Forms.MessageBox.Show("Exception: " + ex.ToString());
    }
}

メソッド BlockBlobClient.openInputStream() クエリアクセラレーション API にクエリを送信し、他の InputStream オブジェクトと同様に読み取ることができる InputStream オブジェクトとして結果をアプリケーションにストリーミングします。

static void QueryHemingway(BlobClient blobClient) {
    String expression = "SELECT * FROM BlobStorage WHERE _3 = 'Hemingway, Ernest, 1899-1961'";
    DumpQueryCsv(blobClient, expression, true);
}

static void DumpQueryCsv(BlobClient blobClient, String query, Boolean headers) {
    try {
        BlobQuerySerialization input = new BlobQueryDelimitedSerialization()
            .setRecordSeparator('\n')
            .setColumnSeparator(',')
            .setHeadersPresent(headers)
            .setFieldQuote('\0')
            .setEscapeChar('\\');
        BlobQuerySerialization output = new BlobQueryDelimitedSerialization()
            .setRecordSeparator('\n')
            .setColumnSeparator(',')
            .setHeadersPresent(true)
            .setFieldQuote('\0')
            .setEscapeChar('\n');
        Consumer<BlobQueryError> errorConsumer = System.out::println;
        Consumer<BlobQueryProgress> progressConsumer = progress -> System.out.println("total bytes read: " + progress.getBytesScanned());
        BlobQueryOptions queryOptions = new BlobQueryOptions(query)
            .setInputSerialization(input)
            .setOutputSerialization(output)
            .setErrorConsumer(errorConsumer)
            .setProgressConsumer(progressConsumer);

        /* Open the query input stream. */
        InputStream stream = blobClient.openQueryInputStream(queryOptions).getValue();
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(stream))) {
            /* Read from stream like you normally would. */
            for (CSVRecord record : CSVParser.parse(reader, CSVFormat.EXCEL.withHeader())) {
                System.out.println(record.toString());
            }
        }
    } catch (Exception e) {
        System.err.println("Exception: " + e.toString());
        e.printStackTrace(System.err);
    }
}

def query_hemingway(blob: BlobClient):
    query = "SELECT * FROM BlobStorage WHERE _3 = 'Hemingway, Ernest, 1899-1961'"
    dump_query_csv(blob, query, False)

def dump_query_csv(blob: BlobClient, query: str, headers: bool):
    qa_reader = blob.query_blob(query, blob_format=DelimitedTextDialect(has_header=headers), on_error=report_error, encoding='utf-8')
    # records() returns a generator that will stream results as received. It will not block pending all results.
    csv_reader = csv.reader(qa_reader.records())
    for row in csv_reader:
        print("*".join(row))

この例では、クエリをクエリアクセラレーション API に送信し、結果をストリームバックします。 queryHemingway ヘルパー関数に渡されるblob オブジェクトの型は BlockBlobClient です。 BlockBlobClient オブジェクトを取得する方法の詳細については、「クイックスタート: Node.jsで JavaScript v12 SDK を使用して BLOB を管理する」を参照してください。

async function queryHemingway(blob)
{
    const query = "SELECT * FROM BlobStorage WHERE _3 = 'Hemingway, Ernest, 1899-1961'";
    await dumpQueryCsv(blob, query, false);
}

async function dumpQueryCsv(blob, query, headers)
{
    var response = await blob.query(query, {
        inputTextConfiguration: {
            kind: "csv",
            recordSeparator: '\n',
            hasHeaders: headers
        },
        outputTextConfiguration: {
            kind: "csv",
            recordSeparator: '\n',
            hasHeaders: true
        },
        onProgress: (progress) => console.log(`Data read: ${progress.loadedBytes}`),
        onError: (err) => console.error(`Error: ${err.position}:${err.name}:${err.description}`)});
    return new Promise(
        function (resolve, reject) {
            csv.parseStream(response.readableStreamBody)
                .on('data', row => console.log(row))
                .on('error', error => {
                    console.error(error);
                    reject(error);
                })
                .on('end', rowCount => resolve());
    });
}

特定の列を取得する

結果の範囲を列のサブセットに設定できます。こうすることで、特定の計算を実行するために必要な列のみを取得できます。これにより、ネットワーク経由で転送されるデータが少なくなるため、アプリケーションのパフォーマンスが向上し、コストが削減されます。

注

結果の範囲を指定できる列の最大数は 49 です。結果に 49 を超える列を含める必要がある場合は、SELECT 式にワイルドカード文字 (*) を使用します (例: SELECT *)。

このコードは、データセット内のすべての書籍の BibNum 列のみを取得します。また、ソースファイルのヘッダー行の情報を使用して、クエリ内の列を参照します。

Function Get-QueryCsv($ctx, $container, $blob, $query, $hasheaders) {
    $tempfile = New-TemporaryFile
    $informat = New-AzStorageBlobQueryConfig -AsCsv -HasHeader:$hasheaders
    Get-AzStorageBlobQueryResult -Context $ctx -Container $container -Blob $blob -InputTextConfiguration $informat -OutputTextConfiguration (New-AzStorageBlobQueryConfig -AsCsv -HasHeader) -ResultFile $tempfile.FullName -QueryString $query -Force
    Get-Content $tempfile.FullName
}

$container = "data"
$blob = "csv/csv-general/seattle-library-with-headers.csv"
Get-QueryCsv $ctx $container $blob "SELECT BibNum FROM BlobStorage" $true

static async Task QueryBibNum(BlockBlobClient blob)
{
    string query = @"SELECT BibNum FROM BlobStorage";
    await DumpQueryCsv(blob, query, true);
}

static void QueryBibNum(BlobClient blobClient)
{
    String expression = "SELECT BibNum FROM BlobStorage";
    DumpQueryCsv(blobClient, expression, true);
}

def query_bibnum(blob: BlobClient):
    query = "SELECT BibNum FROM BlobStorage"
    dump_query_csv(blob, query, True)

async function queryBibNum(blob)
{
    const query = "SELECT BibNum FROM BlobStorage";
    await dumpQueryCsv(blob, query, true);
}

次のコードでは、行のフィルター処理と列プロジェクションを同じクエリに結合します。

Get-QueryCsv $ctx $container $blob $query $true

Function Get-QueryCsv($ctx, $container, $blob, $query, $hasheaders) {
    $tempfile = New-TemporaryFile
    $informat = New-AzStorageBlobQueryConfig -AsCsv -HasHeader:$hasheaders
    Get-AzStorageBlobQueryResult -Context $ctx -Container $container -Blob $blob -InputTextConfiguration $informat -OutputTextConfiguration (New-AzStorageBlobQueryConfig -AsCsv -HasHeader) -ResultFile $tempfile.FullName -QueryString $query -Force
    Get-Content $tempfile.FullName
}

$container = "data"
$query = "SELECT BibNum, Title, Author, ISBN, Publisher, ItemType
            FROM BlobStorage
            WHERE ItemType IN
                ('acdvd', 'cadvd', 'cadvdnf', 'calndvd', 'ccdvd', 'ccdvdnf', 'jcdvd', 'nadvd', 'nadvdnf', 'nalndvd', 'ncdvd', 'ncdvdnf')"

static async Task QueryDvds(BlockBlobClient blob)
{
    string query = @"SELECT BibNum, Title, Author, ISBN, Publisher, ItemType
        FROM BlobStorage
        WHERE ItemType IN
            ('acdvd', 'cadvd', 'cadvdnf', 'calndvd', 'ccdvd', 'ccdvdnf', 'jcdvd', 'nadvd', 'nadvdnf', 'nalndvd', 'ncdvd', 'ncdvdnf')";
    await DumpQueryCsv(blob, query, true);
}

static void QueryDvds(BlobClient blobClient)
{
    String expression = "SELECT BibNum, Title, Author, ISBN, Publisher, ItemType " +
                        "FROM BlobStorage " +
                        "WHERE ItemType IN " +
                        "   ('acdvd', 'cadvd', 'cadvdnf', 'calndvd', 'ccdvd', 'ccdvdnf', 'jcdvd', 'nadvd', 'nadvdnf', 'nalndvd', 'ncdvd', 'ncdvdnf')";
    DumpQueryCsv(blobClient, expression, true);
}

def query_dvds(blob: BlobClient):
    query = "SELECT BibNum, Title, Author, ISBN, Publisher, ItemType "\
        "FROM BlobStorage "\
        "WHERE ItemType IN "\
        "   ('acdvd', 'cadvd', 'cadvdnf', 'calndvd', 'ccdvd', 'ccdvdnf', 'jcdvd', 'nadvd', 'nadvdnf', 'nalndvd', 'ncdvd', 'ncdvdnf')"
    dump_query_csv(blob, query, True)

async function queryDvds(blob)
{
    const query = "SELECT BibNum, Title, Author, ISBN, Publisher, ItemType " +
                  "FROM BlobStorage " +
                  "WHERE ItemType IN " +
                  " ('acdvd', 'cadvd', 'cadvdnf', 'calndvd', 'ccdvd', 'ccdvdnf', 'jcdvd', 'nadvd', 'nadvdnf', 'nalndvd', 'ncdvd', 'ncdvdnf')";
    await dumpQueryCsv(blob, query, true);
}

次の方法で共有

Azure Data Lake Storage クエリ アクセラレーションを使用してデータをフィルター処理する