Filtrare i dati usando l'accelerazione delle query di Azure Data Lake Archiviazione

Articolo
05/30/2024

Questo articolo illustra come usare l'accelerazione delle query per recuperare un subset di dati dall'account di archiviazione.

L'accelerazione delle query consente alle applicazioni e ai framework di analisi di ottimizzare notevolmente l'elaborazione dei dati recuperando solo i dati necessari per eseguire una determinata operazione. Per altre informazioni, vedere Accelerazione query di Azure Data Lake Archiviazione.

Prerequisiti

Per accedere ad Archiviazione di Azure è necessaria una sottoscrizione di Azure. Se non si ha già una sottoscrizione, creare un account gratuito prima di iniziare.
Un account di archiviazione per utilizzo generico v2 . vedere Creare un account di archiviazione.
La crittografia doppia non è supportata.
Se si esegue una query su un file JSON, ogni dimensione del record in questo file deve essere inferiore a 1 MB.
Scegliere una scheda per visualizzare i prerequisiti specifici dell'SDK.
Non applicabile

.NET SDK
- Java Development Kit (JDK) versione 8 o successiva
- Apache Maven
  
  Nota
  
  Questo articolo presuppone che sia stato creato un progetto Java usando Apache Maven. Per un esempio di come creare un progetto usando Apache Maven, vedere Configurazione.
Python 3.8 o versione successiva.

Non sono necessari prerequisiti aggiuntivi per usare Node.js SDK.

Configurazione dell'ambiente

Passaggio 1: Installare i pacchetti

Installare il modulo Az versione 4.6.0 o successiva.

Install-Module -Name Az -Repository PSGallery -Force

Per eseguire l'aggiornamento da una versione precedente di Az, eseguire il comando seguente:

Update-Module -Name Az

Aprire un prompt dei comandi e modificare la directory (cd) nella cartella del progetto, ad esempio:
```
cd myProject
```
Installare la 12.5.0-preview.6 versione o successiva della libreria client di Archiviazione BLOB di Azure per il pacchetto .NET usando il dotnet add package comando .
```
dotnet add package Azure.Storage.Blobs -v 12.8.0
```
Gli esempi visualizzati in questo articolo analizzano un file CSV usando la libreria CsvHelper . Per usare tale libreria, usare il comando seguente.
```
dotnet add package CsvHelper
```

Aprire il file pom.xml del progetto in un editor di testo. Aggiungere gli elementi di dipendenza seguenti al gruppo di dipendenze.

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

Installare la libreria client di Azure Data Lake Archiviazione per Python usando pip.

pip install azure-storage-blob==12.4.0

Installare la libreria client di Data Lake per JavaScript aprendo una finestra del terminale e quindi digitando il comando seguente.

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

Passaggio 2: Aggiungere istruzioni

Non applicabile

Aggiungere queste istruzioni using all'inizio del file di codice.

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

L'accelerazione delle query recupera i dati in formato CSV e Json. Assicurarsi quindi di aggiungere istruzioni using per qualsiasi libreria di analisi CSV o Json che si sceglie di usare. Gli esempi visualizzati in questo articolo analizzano un file CSV usando la libreria CsvHelper disponibile in NuGet. Di conseguenza, queste istruzioni verranno aggiunte using all'inizio del file di codice.

using CsvHelper;
using CsvHelper.Configuration;

Per compilare esempi presentati in questo articolo, è anche necessario aggiungere queste using istruzioni.

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

Aggiungere queste istruzioni import all'inizio del file di codice.

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

Aggiungere queste istruzioni import all'inizio del file di codice.

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

Includere il storage-blob modulo inserendo questa istruzione all'inizio del file di codice.

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

L'accelerazione delle query recupera i dati in formato CSV e Json. Assicurarsi quindi di aggiungere istruzioni per qualsiasi modulo di analisi CSV o Json che si sceglie di usare. Gli esempi visualizzati in questo articolo analizzano un file CSV usando il modulo fast-csv . Pertanto, aggiungeremo questa istruzione all'inizio del file di codice.

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

Recuperare i dati usando un filtro

È possibile usare SQL per specificare i predicati del filtro di riga e le proiezioni di colonne in una richiesta di accelerazione della query. Il codice seguente esegue una query su un file CSV nell'archiviazione e restituisce tutte le righe di dati in cui la terza colonna corrisponde al valore Hemingway, Ernest.

Nella query SQL la parola chiave BlobStorage viene usata per indicare il file su cui viene eseguita la query.
I riferimenti alle colonne vengono specificati come _N dove la prima colonna è _1. Se il file di origine contiene una riga di intestazione, è possibile fare riferimento alle colonne in base al nome specificato nella riga di intestazione.

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

Il metodo BlockBlobClient.QueryAsync asincrono invia la query all'API di accelerazione della query e quindi trasmette i risultati all'applicazione come oggetto 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());
    }
}

Il metodo BlockBlobClient.openInputStream() invia la query all'API di accelerazione della query e quindi trasmette i risultati all'applicazione come InputStream oggetto che può essere letto come qualsiasi altro oggetto 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))

Questo esempio invia la query all'API di accelerazione query e quindi trasmette i risultati. L'oggetto blob passato alla queryHemingway funzione helper è di tipo BlockBlobClient. Per altre informazioni su come ottenere un oggetto BlockBlobClient , vedere Avvio rapido: Gestire i BLOB con JavaScript v12 SDK in 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());
    });
}

Recuperare colonne specifiche

È possibile definire l'ambito dei risultati in un subset di colonne. In questo modo si recuperano solo le colonne necessarie per eseguire un determinato calcolo. Ciò migliora le prestazioni dell'applicazione e riduce i costi perché la rete trasferisce meno dati.

Nota

Il numero massimo di colonne a cui è possibile definire l'ambito dei risultati è 49. Se è necessario che i risultati contengano più di 49 colonne, usare un carattere jolly (*) per l'espressione edizione Standard LECT (ad esempio: SELECT *).

Questo codice recupera solo la BibNum colonna per tutti i libri nel set di dati. Usa anche le informazioni della riga di intestazione nel file di origine per fare riferimento alle colonne nella query.

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

Il codice seguente combina i filtri di riga e le proiezioni di colonne nella stessa query.

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

Condividi tramite