Lettura da BigQuery a Dataflow

Questo documento descrive come leggere i dati da BigQuery a Dataflow utilizzando il connettore I/O BigQuery di Apache Beam.

Nota: a seconda del tuo scenario, puoi utilizzare uno dei modelli Dataflow forniti da Google. molti vengono letti da BigQuery.

Panoramica

Il connettore I/O BigQuery supporta due opzioni per la lettura da BigQuery:

• Letture dirette delle tabelle. Questa è l'opzione più veloce, perché utilizza l'API BigQuery Storage Read.
• Job di esportazione. Con questa opzione, BigQuery esegue un job di esportazione che scrive i dati della tabella in Cloud Storage. che legge i dati esportati da Cloud Storage. Questa opzione è meno efficiente perché richiede la fase di esportazione.

I job di esportazione sono l'opzione predefinita. Per specificare le letture dirette, chiama withMethod(Method.DIRECT_READ).

Il connettore serializza i dati della tabella in un PCollection. Ogni elemento in PCollection rappresenta una singola riga della tabella. Il connettore supporta i seguenti metodi di serializzazione:

• Leggere i dati come record in formato Avro. Utilizzando questo metodo, fornisci una funzione che analizza i record Avro in un tipo di dati personalizzato.
• Leggi i dati come oggetti TableRow. Questo metodo è pratico perché non richiede un tipo di dati personalizzato. Tuttavia, generalmente ha prestazioni inferiori rispetto alla lettura dei record in formato Avro.

Parallelismo

Il parallelismo in questo connettore dipende dal metodo di lettura:

• Lettura diretta: il connettore I/O produce un numero dinamico di flussi in base alle dimensioni della richiesta di esportazione. Legge questi flussi in parallelo direttamente da BigQuery.

• Job di esportazione: BigQuery determina il numero di file da scrivere in Cloud Storage. Il numero di file dipende dalla query e dal volume dei dati. Il connettore I/O legge i file esportati in parallelo.

Prestazioni

La seguente tabella mostra le metriche delle prestazioni per varie opzioni di lettura I/O di BigQuery. I carichi di lavoro sono stati eseguiti su un worker e2-standard2 utilizzando l'SDK Apache Beam 2.49.0 per Java. Non hanno utilizzato Runner v2.

100 M di record | 1 kB | 1 colonna	Velocità effettiva (byte)	Velocità effettiva (elementi)
Lettura archiviazione	120 Mbps	88.000 elementi al secondo
Esportazione Avro	105 Mbps	78.000 elementi al secondo
Esportazione JSON	110 Mbps	81.000 elementi al secondo

Queste metriche si basano su semplici pipeline in modalità batch. Hanno lo scopo di confrontare le prestazioni dei connettori I/O e non sono necessariamente rappresentativi di pipeline reali. Le prestazioni della pipeline Dataflow sono complesse e dipendono dal tipo di VM, dai dati elaborati, dalle prestazioni dei sink e delle origini esterne e dal codice utente. Le metriche si basano sull'esecuzione dell'SDK Java e non sono rappresentative delle caratteristiche delle prestazioni degli SDK di altri linguaggi. Per maggiori informazioni, consulta Prestazioni degli IO di Beam.

best practice

• In generale, consigliamo di utilizzare letture delle tabelle dirette (Method.DIRECT_READ). L'API Storage Read è più adatta alle pipeline di dati rispetto ai job di esportazione, perché non richiede il passaggio intermedio di esportazione dei dati.

• Se utilizzi le letture dirette, ti viene addebitato l'utilizzo dell'API Storage Read. Consulta Prezzi dell'estrazione dati nella pagina dei prezzi di BigQuery.

• Non sono previsti costi aggiuntivi per i job di esportazione. Tuttavia, i job di esportazione hanno dei limiti. Per lo spostamento di dati di grandi dimensioni, dove la tempestività è una priorità e il costo è regolabile, sono consigliate letture dirette.

• L'API Storage Read ha limiti di quota. Utilizza le metriche di Google Cloud per monitorare l'utilizzo della quota.

• Quando utilizzi l'API Storage Read, nei log potresti visualizzare errori di scadenza del lease e di timeout della sessione, ad esempio:

  • DEADLINE_EXCEEDED
  • Server Unresponsive
  • StatusCode.FAILED_PRECONDITION details = "there was an error operating on 'projects/<projectID>/locations/<location>/sessions/<sessionID>/streams/<streamID>': session`

  Questi errori possono verificarsi quando un'operazione richiede più tempo del timeout, di solito in pipeline eseguite per più di 6 ore. Per limitare il problema, passa alle esportazioni di file.

• Quando utilizzi l'SDK Java, valuta la possibilità di creare una classe che rappresenti lo schema della tabella BigQuery. Quindi chiama useBeamSchema nella tua pipeline per convertire automaticamente i tipi Row di Apache Beam e quelli di TableRow BigQuery. Per un esempio di classe schema, consulta ExampleModel.java.

Esempi

Gli esempi di codice in questa sezione utilizzano letture delle tabelle dirette.

Per utilizzare invece un job di esportazione, ometti la chiamata a withMethod o specifica Method.EXPORT. Quindi imposta l'opzione pipeline --tempLocation per specificare un bucket Cloud Storage per i file esportati.

Questi esempi di codice presuppongono che la tabella di origine abbia le seguenti colonne:

• name (stringa)
• age (numero intero)

Specificato come file di schema JSON:

[
  {"name":"user_name","type":"STRING","mode":"REQUIRED"},
  {"name":"age","type":"INTEGER","mode":"REQUIRED"}
]

Lettura di record in formato Avro

Per leggere i dati BigQuery in record in formato Avro, utilizza il metodo read(SerializableFunction). Questo metodo prende una funzione definita dall'applicazione che analizza gli oggetti SchemaAndRecord e restituisce un tipo di dati personalizzato. L'output dal connettore è un PCollection del tipo di dati personalizzati.

Il seguente codice legge un oggetto PCollection<MyData> da una tabella BigQuery, dove MyData è una classe definita dall'applicazione.

Java

Per eseguire l'autenticazione in Dataflow, configura Credenziali predefinite dell'applicazione. Per maggiori informazioni, consulta Configurare l'autenticazione per un ambiente di sviluppo locale.

import org.apache.avro.generic.GenericRecord;
import org.apache.avro.util.Utf8;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.DefaultCoder;
import org.apache.beam.sdk.extensions.avro.coders.AvroCoder;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TypedRead;
import org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.values.TypeDescriptor;

public class BigQueryReadAvro {

  // A custom datatype to hold a record from the source table.
  @DefaultCoder(AvroCoder.class)
  public static class MyData {
    public String name;
    public Long age;

    // Function to convert Avro records to MyData instances.
    public static class FromSchemaAndRecord
            implements SerializableFunction<SchemaAndRecord, MyData> {
      @Override public MyData apply(SchemaAndRecord elem) {
        MyData data = new MyData();
        GenericRecord record = elem.getRecord();
        data.name = ((Utf8) record.get("user_name")).toString();
        data.age = (Long) record.get("age");
        return data;
      }
    }
  }

  public static void main(String[] args) {
    // Parse the pipeline options passed into the application. Example:
    //   --projectId=$PROJECT_ID --datasetName=$DATASET_NAME --tableName=$TABLE_NAME
    // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    PipelineOptionsFactory.register(ExamplePipelineOptions.class);
    ExamplePipelineOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(ExamplePipelineOptions.class);

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        // Read table data into Avro records, using an application-defined parsing function.
        .apply(BigQueryIO.read(new MyData.FromSchemaAndRecord())
            .from(String.format("%s:%s.%s",
                options.getProjectId(),
                options.getDatasetName(),
                options.getTableName()))
            .withMethod(TypedRead.Method.DIRECT_READ))
        // The output from the previous step is a PCollection<MyData>.
        .apply(MapElements
            .into(TypeDescriptor.of(MyData.class))
            .via((MyData x) -> {
              System.out.printf("Name: %s, Age: %d%n", x.name, x.age);
              return x;
            }));
    pipeline.run().waitUntilFinish();
  }
}

Il metodo read utilizza un'interfaccia SerializableFunction<SchemaAndRecord, T>, che definisce una funzione per la conversione dai record Avro a una classe di dati personalizzata. Nell'esempio di codice precedente, il metodo MyData.apply implementa questa funzione di conversione. La funzione di esempio analizza i campi name e age dal record Avro e restituisce un'istanza MyData.

Nota: questo codice restituisce un errore se la funzione di analisi non corrisponde allo schema effettivo della tabella BigQuery. In particolare, devono esistere le colonne denominate e i tipi di dati devono essere compatibili. Inoltre, se sono presenti colonne nella tabella nulli, il codice deve gestire valori nulli.

Per specificare la tabella BigQuery da leggere, chiama il metodo from, come mostrato nell'esempio precedente. Per ulteriori informazioni, consulta Nomi delle tabelle nella documentazione del connettore I/O BigQuery.

Leggi TableRow oggetti

Il metodo readTableRows legge i dati di BigQuery in PCollection di TableRow oggetti. Ogni TableRow è una mappa di coppie chiave-valore che contiene una singola riga di dati della tabella. Specifica la tabella BigQuery da leggere chiamando il metodo from.

Il seguente codice legge un oggetto PCollection<TableRows> da una tabella BigQuery.

Java

Per eseguire l'autenticazione in Dataflow, configura Credenziali predefinite dell'applicazione. Per maggiori informazioni, consulta Configurare l'autenticazione per un ambiente di sviluppo locale.

import com.google.api.services.bigquery.model.TableRow;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TypedRead.Method;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptor;

public class BiqQueryReadTableRows {
  public static void main(String[] args) {
    // Parse the pipeline options passed into the application. Example:
    //   --projectId=$PROJECT_ID --datasetName=$DATASET_NAME --tableName=$TABLE_NAME
    // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    PipelineOptionsFactory.register(ExamplePipelineOptions.class);
    ExamplePipelineOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(ExamplePipelineOptions.class);

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        // Read table data into TableRow objects.
        .apply(BigQueryIO.readTableRows()
            .from(String.format("%s:%s.%s",
                options.getProjectId(),
                options.getDatasetName(),
                options.getTableName()))
            .withMethod(Method.DIRECT_READ)
        )
        // The output from the previous step is a PCollection<TableRow>.
        .apply(MapElements
            .into(TypeDescriptor.of(TableRow.class))
            // Use TableRow to access individual fields in the row.
            .via((TableRow row) -> {
              var name = (String) row.get("user_name");
              var age = (String) row.get("age");
              System.out.printf("Name: %s, Age: %s%n", name, age);
              return row;
            }));
    pipeline.run().waitUntilFinish();
  }
}

Questo esempio mostra anche come accedere ai valori dal dizionario TableRow. I valori interi sono codificati come stringhe per corrispondere al formato JSON esportato di BigQuery.

Nota: questo codice restituisce un errore se la funzione di analisi non corrisponde allo schema effettivo della tabella BigQuery. In particolare, devono esistere le colonne denominate e i tipi di dati devono essere compatibili. Inoltre, se sono presenti colonne nella tabella nulli, il codice deve gestire valori nulli.

Proiezione e filtri delle colonne

Quando utilizzi le letture dirette (Method.DIRECT_READ), puoi rendere più efficienti le operazioni di lettura riducendo la quantità di dati letti da BigQuery e inviati sulla rete.

• Proiezione delle colonne: chiama withSelectedFields per leggere un sottoinsieme di colonne della tabella. Ciò consente letture efficienti quando le tabelle contengono molte colonne.
• Filtro delle righe: chiama withRowRestriction per specificare un predicato che filtra i dati sul lato server.

I predicati del filtro devono essere deterministici e l'aggregazione non è supportata.

Nota: questi metodi non sono supportati durante la lettura da un job di esportazione (Method.EXPORT).

L'esempio seguente esegue una previsione delle colonne "user_name" e "age" ed esclude le righe che non corrispondono al predicato "age > 18".

Java

Per eseguire l'autenticazione in Dataflow, configura Credenziali predefinite dell'applicazione. Per maggiori informazioni, consulta Configurare l'autenticazione per un ambiente di sviluppo locale.

import com.google.api.services.bigquery.model.TableRow;
import java.util.Arrays;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TypedRead;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptor;

public class BigQueryReadWithProjectionAndFiltering {
  public static void main(String[] args) {
    // Parse the pipeline options passed into the application. Example:
    //   --projectId=$PROJECT_ID --datasetName=$DATASET_NAME --tableName=$TABLE_NAME
    // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    PipelineOptionsFactory.register(ExamplePipelineOptions.class);
    ExamplePipelineOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(ExamplePipelineOptions.class);

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        .apply(BigQueryIO.readTableRows()
            // Read rows from a specified table.
            .from(String.format("%s:%s.%s",
                options.getProjectId(),
                options.getDatasetName(),
                options.getTableName()))
            .withMethod(TypedRead.Method.DIRECT_READ)
            .withSelectedFields(Arrays.asList("user_name", "age"))
            .withRowRestriction("age > 18")
        )
        // The output from the previous step is a PCollection<TableRow>.
        .apply(MapElements
            .into(TypeDescriptor.of(TableRow.class))
            // Use TableRow to access individual fields in the row.
            .via((TableRow row) -> {
              var name = (String) row.get("user_name");
              var age = row.get("age");
              System.out.printf("Name: %s, Age: %s%n", name, age);
              return row;
            }));
    pipeline.run().waitUntilFinish();
  }
}

Leggere dal risultato di una query

Gli esempi precedenti mostrano come leggere le righe di una tabella. Puoi anche leggere dal risultato di una query SQL, chiamando fromQuery. Questo approccio sposta parte del lavoro di calcolo in BigQuery. Puoi utilizzare questo metodo anche per leggere da una vista BigQuery o da una vista materializzata, eseguendo una query sulla vista.

L'esempio seguente esegue una query su un set di dati pubblico BigQuery e legge i risultati. Dopo l'esecuzione della pipeline, puoi vedere il job di query nella cronologia dei job BigQuery.

Java

Per eseguire l'autenticazione in Dataflow, configura Credenziali predefinite dell'applicazione. Per maggiori informazioni, consulta Configurare l'autenticazione per un ambiente di sviluppo locale.

import com.google.api.services.bigquery.model.TableRow;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TypedRead;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptor;

public class BigQueryReadFromQuery {
  public static void main(String[] args) {
    // The SQL query to run inside BigQuery.
    final String queryString =
        "SELECT repo_name as repo, COUNT(*) as count "
            + "FROM `bigquery-public-data.github_repos.sample_commits` "
            + "GROUP BY repo_name";

    // Parse the pipeline options passed into the application.
    // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    PipelineOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation().create();

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        // Read the query results into TableRow objects.
        .apply(BigQueryIO.readTableRows()
            .fromQuery(queryString)
            .usingStandardSql()
            .withMethod(TypedRead.Method.DIRECT_READ))
        // The output from the previous step is a PCollection<TableRow>.
        .apply(MapElements
            .into(TypeDescriptor.of(TableRow.class))
            .via((TableRow row) -> {
              System.out.printf("Repo: %s, commits: %s%n", row.get("repo"), row.get("count"));
              return row;
            }));
    pipeline.run().waitUntilFinish();
  }
}

Passaggi successivi

• Leggi la documentazione sul connettore BigQuery I/O.
• Consulta l'elenco dei modelli forniti da Google.