Filtrowanie danych przy użyciu przyspieszania zapytań usługi Azure Data Lake Storage

W tym artykule pokazano, jak używać przyspieszenia zapytań do pobierania podzestawu danych z konta przechowywania.

Przyspieszanie zapytań umożliwia aplikacjom i strukturom analitycznym znaczne zoptymalizowanie przetwarzania danych przez pobranie tylko danych, których potrzebują do wykonania danej operacji. Aby dowiedzieć się więcej, zobacz Przyspieszanie zapytań usługi Azure Data Lake Storage.

Wymagania wstępne

• Aby uzyskać dostęp do usługi Azure Storage, potrzebujesz subskrypcji platformy Azure. Jeśli nie masz jeszcze subskrypcji, przed rozpoczęciem utwórz bezpłatne konto .

• Konto magazynowe w wersji 2 ogólnego przeznaczenia. Zobacz Tworzenie konta magazynu.

• Podwójne szyfrowanie nie jest obsługiwane.

• Jeśli wysyłasz zapytanie do pliku JSON, każdy rozmiar rekordu w tym pliku powinien być mniejszy niż 1 MB.

• Wybierz kartę, aby wyświetlić wszystkie wymagania wstępne specyficzne dla zestawu SDK.

  PowerShell

  Nie dotyczy

  .NET

  Zestaw SDK dla platformy .NET

  Java

  • Zestaw Java Development Kit (JDK) w wersji 8 lub nowszej

  • Apache Maven

    Uwaga

    W tym artykule założono, że utworzono projekt Java przy użyciu narzędzia Apache Maven. Aby zapoznać się z przykładem tworzenia projektu przy użyciu narzędzia Apache Maven, zobacz Konfigurowanie.

  Python

  Python 3.8 lub nowszy.

  Node.js

  Nie ma dodatkowych wymagań wstępnych wymaganych do korzystania z zestawu SDK Node.js.

---

Konfigurowanie środowiska

Krok 1. Instalowanie pakietów

PowerShell

Zainstaluj moduł Az w wersji 4.6.0 lub nowszej.

Install-Module -Name Az -Repository PSGallery -Force

Aby przeprowadzić aktualizację ze starszej wersji modułu Az, uruchom następujące polecenie:

Update-Module -Name Az

.NET

1. Otwórz wiersz polecenia i zmień katalog (cd) w folderze projektu, na przykład:

   cd myProject
   

2. Zainstaluj wersję biblioteki klienta usługi Azure Blob Storage dla pakietu .NET 12.5.0-preview.6 lub nowszą, używając polecenia dotnet add package.

   dotnet add package Azure.Storage.Blobs -v 12.8.0
   

3. Przykłady wyświetlane w tym artykule analizują plik CSV przy użyciu biblioteki CsvHelper . Aby użyć tej biblioteki, użyj następującego polecenia.

   dotnet add package CsvHelper
   

Java

1. Otwórz plik pom.xml projektu w edytorze tekstów. Dodaj następujące elementy zależności do grupy zależności.

   <!-- 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>
   

Python

Zainstaluj bibliotekę klienta usługi Azure Data Lake Storage dla języka Python przy użyciu narzędzia pip.

pip install azure-storage-blob==12.4.0

Node.js

Zainstaluj bibliotekę klienta usługi Data Lake dla języka JavaScript, otwierając okno terminalu, a następnie wpisując następujące polecenie.

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

Krok 2: Dodaj instrukcje

PowerShell

Nie dotyczy

.NET

Dodaj te using instrukcje na początku pliku kodu.

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

Przyspieszanie zapytań pobiera dane sformatowane w formacie CSV i JSON. W związku z tym pamiętaj, aby dodać instrukcje using dla dowolnych bibliotek do analizowania plików CSV lub JSON, których chcesz użyć. Przykłady wyświetlane w tym artykule analizują plik CSV przy użyciu biblioteki CsvHelper dostępnej w narzędziu NuGet. W związku z tym dodalibyśmy te using instrukcje na początku pliku kodu.

using CsvHelper;
using CsvHelper.Configuration;

Aby skompilować przykłady przedstawione w tym artykule, należy dodać także te instrukcje using.

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

Java

Dodaj te import instrukcje na początku pliku kodu.

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.*;

Python

Dodaj te instrukcje importu na początku pliku kodu.

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

Node.js

storage-blob Dołącz moduł, umieszczając tę instrukcję w górnej części pliku kodu.

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

Przyspieszanie zapytań pobiera dane sformatowane w formacie CSV i JSON. W związku z tym upewnij się, że dodano instrukcje dla dowolnych modułów analizowania plików CSV lub JSON, których chcesz użyć. Przykłady wyświetlane w tym artykule analizują plik CSV przy użyciu modułu fast-csv . W związku z tym dodalibyśmy tę instrukcję na początku pliku kodu.

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

Pobieranie danych przy użyciu filtru

Za pomocą języka SQL można określić predykaty filtru wierszy i projekcje kolumn w żądaniu przyspieszania zapytań. Poniższy kod wysyła zapytanie do pliku CSV w magazynie i zwraca wszystkie wiersze danych, w których trzecia kolumna odpowiada wartości Hemingway, Ernest.

• W zapytaniu SQL słowo kluczowe BlobStorage jest używane do oznaczania pliku, którego dotyczy zapytanie.

• Odwołania do kolumn są podawane jako _N, gdzie pierwsza kolumna to _1. Jeśli plik źródłowy zawiera wiersz nagłówka, możesz odwołać się do kolumn według nazwy określonej w wierszu nagłówka.

PowerShell

Function Get-QueryCsv($ctx, $container, $blob, $query, $hasheaders) {
    $tempfile = New-TemporaryFile
    $informat = New-AzStorageBlobQueryConfig -AsCsv -HasHeader:$hasheaders -RecordSeparator "`n" -ColumnSeparator "," -QuotationCharacter """" -EscapeCharacter "\"
    Get-AzStorageBlobQueryResult -Context $ctx -Container $container -Blob $blob -InputTextConfiguration $informat -OutputTextConfiguration (New-AzStorageBlobQueryConfig -AsCsv -HasHeader -RecordSeparator "`n" -ColumnSeparator "," -QuotationCharacter """" -EscapeCharacter "\") -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

.NET

Metoda asynchroniczna BlockBlobClient.QueryAsync wysyła zapytanie do interfejsu API przyspieszania zapytań, a następnie przesyła strumieniowo wyniki z powrotem do aplikacji jako obiekt 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());
    }
}

Java

Metoda BlockBlobClient.openInputStream() wysyła zapytanie do interfejsu API przyspieszania zapytań, a następnie przesyła strumieniowo wyniki z powrotem do aplikacji jako InputStream obiekt, który może być odczytywany jak każdy inny obiekt 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);
    }
}

Python

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))

Node.js

Ten przykład wysyła zapytanie do API dla przyspieszonego przetwarzania zapytań, a następnie przesyła strumieniowo wyniki. Obiekt blob przekazany do funkcji pomocniczej queryHemingway jest typu BlockBlobClient. Aby dowiedzieć się więcej o sposobie uzyskiwania obiektu BlockBlobClient , zobacz Szybki start: zarządzanie obiektami blob przy użyciu zestawu SDK języka JavaScript w wersji 12 w Node.js.

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());
    });
}

Pobieranie określonych kolumn

Wyniki można ograniczyć do podzbioru kolumn. W ten sposób pobierasz tylko kolumny potrzebne do wykonania danego obliczenia. Zwiększa to wydajność aplikacji i zmniejsza koszty, ponieważ mniej danych jest przesyłanych za pośrednictwem sieci.

Uwaga

Maksymalna liczba kolumn, do których można ograniczyć zakres wyników, wynosi 49. Jeśli wyniki mają zawierać więcej niż 49 kolumn, użyj symbolu wieloznakowego (*) dla wyrażenia SELECT (na przykład: SELECT *).

Ten kod pobiera tylko kolumnę BibNum dla wszystkich książek w zestawie danych. Używa również informacji z wiersza nagłówka w pliku źródłowym do odwoływania się do kolumn w zapytaniu.

PowerShell

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

.NET

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

Java

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

Python

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

Node.js

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

Poniższy kod łączy filtrowanie wierszy i projekcje kolumn w tym samym zapytaniu.

PowerShell

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')"

.NET

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);
}

Java

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);
}

Python

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)

Node.js

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);
}

Następne kroki

• Przyspieszanie zapytań usługi Azure Data Lake Storage
• Dokumentacja języka SQL przyspieszania zapytań