Filtrer les données à l’aide de l’accélération des requêtes Azure Data Lake Storage

Cet article explique comment utiliser l’accélération des requêtes pour récupérer un sous-ensemble de données à partir de votre compte de stockage.

L’accélération des requêtes permet aux applications et aux infrastructures d’analytique d’optimiser considérablement le traitement des données en récupérant uniquement les données dont elles ont besoin pour effectuer une opération donnée. Pour plus d’informations, consultez Accélération des requêtes Azure Data Lake Storage.

Conditions préalables

• Pour accéder à Stockage Azure, vous avez besoin d’un abonnement Azure. Si vous n’avez pas d’abonnement, vous pouvez créer un compte gratuit avant de commencer.

• Un compte de stockage v2 universel . consultez Créer un compte de stockage.

• Le double chiffrement n’est pas pris en charge.

• Si vous interrogez un fichier JSON, chaque taille d’enregistrement de ce fichier doit être inférieure à 1 Mo.

• Choisissez un onglet pour afficher les prérequis spécifiques au Kit de développement logiciel (SDK).

  PowerShell

  Sans objet

  .NET

  Kit de développement logiciel (SDK) .NET

  Java

  • Kit de développement Java (JDK), version 8 ou ultérieure

  • Apache Maven

    Remarque

    Cet article part du principe que vous avez créé un projet Java à l’aide d’Apache Maven. Pour obtenir un exemple de création d’un projet à l’aide d’Apache Maven, consultez Configuration.

  Python

  Python 3.8 ou version ultérieure.

  Node.JS

  Il n’existe aucun prérequis supplémentaire requis pour utiliser le KIT DE développement logiciel (SDK) Node.js.

---

Configurer votre environnement

Étape 1 : Installer des packages

PowerShell

Installez le module Az version 4.6.0 ou ultérieure.

Install-Module -Name Az -Repository PSGallery -Force

Pour effectuer une mise à jour à partir d’une version antérieure d’Az, exécutez la commande suivante :

Update-Module -Name Az

.NET

1. Ouvrez une invite de commandes et modifiez le répertoire (cd) dans votre dossier de projet, par exemple :

   cd myProject
   

2. Installez la 12.5.0-preview.6 version ou une version ultérieure de la bibliothèque de client stockage Blob Azure pour le package .NET à l’aide de la dotnet add package commande.

   dotnet add package Azure.Storage.Blobs -v 12.8.0
   

3. Les exemples qui apparaissent dans cet article analysent un fichier CSV à l’aide de la bibliothèque CsvHelper . Pour utiliser cette bibliothèque, utilisez la commande suivante.

   dotnet add package CsvHelper
   

Java

1. Ouvrez le fichier pom.xml de votre projet dans un éditeur de texte. Ajoutez les éléments de dépendance suivants au groupe de dépendances.

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

Installez la bibliothèque cliente Azure Data Lake Storage pour Python à l’aide de pip.

pip install azure-storage-blob==12.4.0

Node.js

Installez la bibliothèque de client Data Lake pour JavaScript en ouvrant une fenêtre de terminal, puis en entrant la commande suivante.

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

Étape 2 : Ajouter des déclarations

PowerShell

Sans objet

.NET

Ajoutez ces using instructions en haut de votre fichier de code.

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

L’accélération des requêtes récupère les données au format CSV et JSON. Par conséquent, veillez à ajouter des instructions using pour toutes les bibliothèques d’analyse CSV ou JSON que vous choisissez d’utiliser. Les exemples qui apparaissent dans cet article analysent un fichier CSV à l’aide de la bibliothèque CsvHelper disponible sur NuGet. Par conséquent, nous allons ajouter ces using instructions en haut du fichier de code.

using CsvHelper;
using CsvHelper.Configuration;

Pour compiler des exemples présentés dans cet article, vous devez également ajouter ces using instructions.

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

Java

Ajoutez ces import instructions en haut de votre fichier de code.

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

Ajoutez ces instructions d’importation en haut de votre fichier de code.

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

Node.JS

Incluez le storage-blob module en plaçant cette instruction en haut de votre fichier de code.

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

L’accélération des requêtes récupère les données au format CSV et JSON. Par conséquent, veillez à ajouter des instructions pour tous les modules d’analyse CSV ou JSON que vous choisissez d’utiliser. Les exemples qui apparaissent dans cet article analysent un fichier CSV à l’aide du module fast-csv . Par conséquent, nous allons ajouter cette instruction en haut du fichier de code.

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

Récupérer des données à l’aide d’un filtre

Vous pouvez utiliser SQL pour spécifier les prédicats de filtre de lignes et les projections de colonnes dans une requête d’accélération de requête. Le code suivant interroge un fichier CSV dans le stockage et retourne toutes les lignes de données où la troisième colonne correspond à la valeur Hemingway, Ernest.

• Dans la requête SQL, le mot clé BlobStorage est utilisé pour indiquer le fichier interrogé.

• Les références de colonne sont spécifiées comme _N où la première colonne est _1. Si le fichier source contient une ligne d’en-tête, vous pouvez faire référence aux colonnes par le nom spécifié dans la ligne d’en-tête.

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

La méthode BlockBlobClient.QueryAsync asynchrone envoie la requête à l’API d’accélération de requête, puis transmet les résultats à l’application en tant qu’objet 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

La méthode BlockBlobClient.openInputStream() envoie la requête à l’API d’accélération de requête, puis transmet les résultats à l’application en tant qu’objet InputStream qui peut être lu comme n’importe quel autre objet 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

Cet exemple montre comment envoyer la requête à l’API d’accélération de requête, puis renvoyer les résultats. L’objet blob passé dans la queryHemingway fonction d’assistance est de type BlockBlobClient. Pour en savoir plus sur l’obtention d’un objet BlockBlobClient , consultez Démarrage rapide : Gérer des objets blob avec le Kit de développement logiciel (SDK) JavaScript v12 dans 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());
    });
}

Récupérer des colonnes spécifiques

Vous pouvez étendre vos résultats à un sous-ensemble de colonnes. De cette façon, vous récupérez uniquement les colonnes nécessaires pour effectuer un calcul donné. Cela améliore les performances des applications et réduit les coûts, car moins de données sont transférées sur le réseau.

Remarque

Le nombre maximal de colonnes que vous pouvez étendre à vos résultats est de 49. Si vous avez besoin que vos résultats contiennent plus de 49 colonnes, utilisez un caractère générique (*) pour l’expression SELECT (par exemple : SELECT *).

Ce code récupère uniquement la BibNum colonne de tous les livres du jeu de données. Il utilise également les informations de la ligne d’en-tête dans le fichier source pour référencer des colonnes dans la requête.

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

Le code suivant combine le filtrage de lignes et les projections de colonnes dans la même requête.

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

Étapes suivantes

• Accélération des requêtes Azure Data Lake Storage
• Informations de référence sur le langage SQL d’accélération des requêtes