Filtro de datos mediante la aceleración de consultas de Azure Data Lake Storage

Artículo
06/01/2023

En este artículo se muestra cómo usar la aceleración de consultas para recuperar un subconjunto de datos de la cuenta de almacenamiento.

La aceleración de consultas permite tanto a las aplicaciones como a los marcos de análisis optimizar considerablemente el procesamiento de datos mediante la recuperación solo de los datos necesarios para realizar una operación determinada. Para más información, consulte Aceleración de consultas de Azure Data Lake Storage.

Requisitos previos

Para acceder a Azure Storage, necesitará una suscripción de Azure. Si todavía no tiene una suscripción, cree una cuenta gratuita antes de empezar.
Una cuenta de almacenamiento de uso general v2. Vea Creación de una cuenta de Azure Storage.
No se admite el cifrado doble.
Si está consultando un archivo JSON, cada tamaño de registro de este archivo debe ser menor que 1 MB.
Elija una pestaña para ver todos los requisitos previos específicos del SDK.
No aplicable

El SDK de .NET
- Kit de desarrollo de Java (JDK), versión 8 o posterior
- Apache Maven
  
  Nota:
  
  En este artículo se considera que ha creado un proyecto de Java mediante Apache Maven. Para obtener un ejemplo de cómo crear un proyecto con Apache Maven, consulte Instalación.
Python3.8 o superior.

No hay requisitos previos adicionales necesarios para usar el SDK de Node.js.

Configurar el entorno

Paso 1: Instalar paquetes

Instale la versión 4.6.0 o superior del módulo Az.

Install-Module -Name Az -Repository PSGallery -Force

Para actualizar desde una versión anterior de Az, ejecute el siguiente comando:

Update-Module -Name Az

Abra un símbolo del sistema y cambie el directorio (cd) a la carpeta del proyecto, por ejemplo:
```
cd myProject
```
Instale la versión 12.5.0-preview.6 o posterior de biblioteca cliente de Azure Blob Storage para el paquete .NET mediante el comando dotnet add package.
```
dotnet add package Azure.Storage.Blobs -v 12.8.0
```
En los ejemplos que aparecen en este artículo se analiza un archivo CSV mediante la biblioteca de CsvHelper. Para usar esa biblioteca, use el siguiente comando.
```
dotnet add package CsvHelper
```

Abra el archivo pom.xml del proyecto en un editor de texto. Agregue los siguientes elementos de dependencia al grupo de dependencias.

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

Instale la biblioteca de cliente de Azure Data Lake Storage para Python con PIP.

pip install azure-storage-blob==12.4.0

Instale la biblioteca cliente de Data Lake para JavaScript; para ello, abra una ventana de terminal y, a continuación, escriba el siguiente comando.

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

Paso 2: Adición de instrucciones

No aplicable

Agregue estas instrucciones using al inicio del archivo de código.

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

La aceleración de consultas recupera datos con formato CSV y JSON. Por lo tanto, asegúrese de agregar instrucciones USING para las bibliotecas de análisis de CSV o JSON que decida usar. En los ejemplos que aparecen en este artículo se analiza un archivo CSV mediante la biblioteca de CsvHelper que está disponible en NuGet. Por lo tanto, se agregan estas instrucciones using al inicio del archivo de código.

using CsvHelper;
using CsvHelper.Configuration;

Para compilar los ejemplos presentados en este artículo, también deberá agregar estas instrucciones using.

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

Agregue estas instrucciones import al inicio del archivo de código.

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

Agregue estas instrucciones import al inicio del archivo de código.

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

Incluya el módulo storage-blob colocando esta instrucción al principio del archivo del código.

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

La aceleración de consultas recupera datos con formato CSV y JSON. Por consiguiente, asegúrese de agregar instrucciones para todos los módulo de análisis de CSV o JSON que decida usar. En los ejemplos que aparecen en este artículo se analiza un archivo CSV mediante el módulo fast-csv. Por consiguiente, agregaremos esta instrucción al principio del archivo de código.

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

Recuperación de datos mediante un filtro

Puede usar SQL para especificar los predicados de filtro de fila y las proyecciones de columna en una solicitud de aceleración de consultas. En el código siguiente se consulta un archivo CSV en el almacenamiento y se devuelven todas las filas de datos en las que la tercera columna coincide con el valor Hemingway, Ernest.

En la consulta SQL, se usa la palabra clave BlobStorage para indicar el archivo que se consulta.
Las referencias de columna se especifican como _N, donde la primera columna es _1. Si el archivo de código fuente contiene una fila de encabezado, puede hacer referencia a las columnas por el nombre especificado en la fila de encabezado.

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

El método asincrónico BlobQuickQueryClient.QueryAsync envía la consulta a la API de aceleración de consultas y, a continuación, vuelve a transmitir los resultados a la aplicación como un objeto 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());
    }
}

El método BlobQuickQueryClient.openInputStream() envía la consulta a la API de aceleración de consultas y, a continuación, vuelve a transmitir los resultados a la aplicación como un objeto InputStream que se puede leer como cualquier otro objeto 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))

En este ejemplo se envía la consulta a la API de aceleración de consultas y luego se hace streaming de los resultados. El objeto blob que se pasa a la función auxiliar queryHemingway es del tipo BlockBlobClient. Para obtener más información acerca de cómo obtener un objeto BlockBlobClient, consulte Inicio rápido: Administración de blobs con el SDK de JavaScript v12 en 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());
    });
}

Recuperación de columnas específicas

Puede limitar los resultados a un subconjunto de columnas. De este modo, solo recuperará las columnas necesarias para realizar un cálculo determinado. Esto mejora el rendimiento de la aplicación y reduce los costos porque se transfieren menos datos a través de la red.

Nota:

El número máximo de columnas a las que se pueden ampliar los resultados es 49. Si necesita que los resultados contengan más de 49 columnas, use un carácter comodín (*) para la expresión SELECT (por ejemplo: SELECT *).

Este código solo recupera la columna BibNum para todos los libros del conjunto de datos. También utiliza la información de la fila de encabezado del archivo de origen para hacer referencia a las columnas de la consulta.

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

El código siguiente combina el filtrado de filas y las proyecciones de columnas en la misma consulta.

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