Lee de BigQuery a Dataflow

En este documento, se describe cómo leer datos de BigQuery a Dataflow.

Descripción general

Para la mayoría de los casos de uso, considera usar la E/S administrada para leer desde BigQuery. El E/S administrado proporciona funciones como actualizaciones automáticas y una API de configuración coherente. Cuando se lee desde BigQuery, la E/S administrada realiza lecturas directas de tablas, lo que ofrece el mejor rendimiento de lectura.

Si necesitas un ajuste de rendimiento más avanzado, considera usar el conector BigQueryIO. El conector BigQueryIO admite la lectura directa de tablas y la lectura de trabajos de exportación de BigQuery. También ofrece un control más detallado sobre la desserialización de los registros de tablas. Para obtener más información, consulta Usa el conector BigQueryIO en este documento.

Proyección y filtrado de columnas

Para reducir el volumen de datos que lee tu canalización de BigQuery, puedes usar las siguientes técnicas:

• La proyección de columnas especifica un subconjunto de columnas para leer de la tabla. Usa la proyección de columnas cuando tu tabla tenga una gran cantidad de columnas y solo necesites leer un subconjunto de ellas.
• El filtrado de filas especifica un predicado para aplicar a la tabla. La operación de lectura de BigQuery solo muestra filas que coinciden con el filtro, lo que puede reducir la cantidad total de datos que transfiere la canalización.

En el siguiente ejemplo, se leen las columnas "user_name" y "age" de una tabla y se filtran las filas que no coinciden con el predicado "age > 18". En este ejemplo, se usa la E/S administrada.

Java

Para autenticarte en Dataflow, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para un entorno de desarrollo local.

import com.google.common.collect.ImmutableMap;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.managed.Managed;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.Row;
import org.apache.beam.sdk.values.TypeDescriptors;

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

    String tableSpec = String.format("%s:%s.%s",
        options.getProjectId(),
        options.getDatasetName(),
        options.getTableName());

    ImmutableMap<String, Object> config = ImmutableMap.<String, Object>builder()
        .put("table", tableSpec)
        .put("row_restriction", "age > 18")
        .put("fields", List.of("user_name", "age"))
        .build();

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        .apply(Managed.read(Managed.BIGQUERY).withConfig(config)).getSinglePCollection()
        .apply(MapElements
            .into(TypeDescriptors.strings())
            // Access individual fields in the row.
            .via((Row row) -> {
              String output = String.format("Name: %s, Age: %s%n",
                  row.getString("user_name"),
                  row.getInt64("age"));
              System.out.println(output);
              return output;
            }));
    pipeline.run().waitUntilFinish();
  }
}

Lee desde el resultado de una consulta

En el siguiente ejemplo, se usa E/S administrada para leer el resultado de una consulta de SQL. Ejecuta una consulta en un conjunto de datos públicos de BigQuery. También puedes usar consultas de SQL para leer desde una vista de BigQuery o una vista materializada.

Java

Para autenticarte en Dataflow, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para un entorno de desarrollo local.

import com.google.common.collect.ImmutableMap;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.managed.Managed;
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.Row;
import org.apache.beam.sdk.values.TypeDescriptors;

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

    ImmutableMap<String, Object> config = ImmutableMap.<String, Object>builder()
        .put("query", queryString)
        .build();

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        .apply(Managed.read(Managed.BIGQUERY).withConfig(config)).getSinglePCollection()
        .apply(MapElements
            .into(TypeDescriptors.strings())
            // Access individual fields in the row.
            .via((Row row) -> {
              String output = String.format("Repo: %s, commits: %d%n",
                  row.getString("repo"),
                  row.getInt64("count"));
              System.out.println(output);
              return output;
            }));
    pipeline.run().waitUntilFinish();
  }
}

Usa el conector BigQueryIO

El conector BigQueryIO admite los siguientes métodos de serialización:

• Lee los datos como registros con formato Avro. Con este método, proporcionas una función que analiza los registros de Avro en un tipo de datos personalizado.
• Lee los datos como objetos TableRow. Este método es conveniente porque no requiere un tipo de datos personalizado. Sin embargo, suele tener un rendimiento menor que leer los registros con formato Avro.

El conector admite dos opciones para leer datos:

• Exportar trabajo. De forma predeterminada, el conector BigQueryIO ejecuta un trabajo de exportación de BigQuery que escribe los datos de la tabla en Cloud Storage. Luego, el conector lee los datos de Cloud Storage.
• Lecturas directas de tablas: Esta opción es más rápida que los trabajos de exportación, ya que usa la API de BigQuery Storage Read y omite el paso de exportación. Para usar lecturas de tablas directas, llama a withMethod(Method.DIRECT_READ) cuando compiles la canalización.

Cuando elijas qué opción usar, ten en cuenta los siguientes puntos:

• En general, recomendamos usar lecturas de tablas directas. La API de Storage Read es más adecuada para canalizaciones de datos que para trabajos de exportación, ya que no necesita el paso intermedio de exportar datos.

• Si usas lecturas directas, se te cobra por el uso de la API de Storage Read. Consulta Precios de extracción de datos en la página de precios de BigQuery.

• Los trabajos de exportación no tienen costo adicional. Sin embargo, los trabajos de exportación tienen límites. Para movimientos de datos grandes, en los que la puntualidad es una prioridad y el costo es ajustable, se recomiendan las lecturas directas.

• La API de Storage Read tiene límites de cuota. Usa las métricas deGoogle Cloud para supervisar el uso de la cuota.

• Si usas trabajos de exportación, configura la opción de canalización --tempLocation para especificar un bucket de Cloud Storage para los archivos exportados.

• Cuando usas la API de lectura de almacenamiento, es posible que veas errores de vencimiento y tiempo de espera de sesión en los registros, como los siguientes:

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

  Estos errores pueden ocurrir cuando una operación tarda más que el tiempo de espera, por lo general, en canalización que se ejecutan durante más de 6 horas. Para mitigar este problema, cambia a las exportaciones de archivos.

• El grado de paralelismo depende del método de lectura:

  • Lecturas directas: el conector de E/S produce una cantidad dinámica de transmisiones, según el tamaño de la solicitud de exportación. Lee estas transmisiones directamente desde BigQuery en paralelo.

  • Trabajos de exportación: BigQuery determina cuántos archivos se escriben en Cloud Storage. La cantidad de archivos depende de la consulta y del volumen de datos. El conector de E/S lee los archivos exportados en paralelo.

En la siguiente tabla, se muestran las métricas de rendimiento de varias opciones de lectura de E/S de BigQuery. Las cargas de trabajo se ejecutaron en un trabajador e2-standard2, con el SDK de Apache Beam 2.49.0 para Java. No usaron Runner v2.

100 millones de registros | 1 KB | 1 columna	Capacidad de procesamiento (bytes)	Capacidad de procesamiento (elementos)
Lectura de almacenamiento	120 MBps	88,000 elementos por segundo
Exportación de Avro	105 MBps	78,000 elementos por segundo
Exportación de Json	110 MBps	81,000 elementos por segundo

Estas métricas se basan en canalizaciones por lotes simples. Están diseñadas para comparar el rendimiento entre los conectores de E/S y no representan necesariamente las canalizaciones del mundo real. El rendimiento de la canalización de Dataflow es complejo y es una función del tipo de VM, los datos que se procesan, el rendimiento de las fuentes y los receptores externos y el código de usuario. Las métricas se basan en la ejecución del SDK de Java y no representan las características de rendimiento de otros SDK de lenguaje. Para obtener más información, consulta Rendimiento de E/S de Beam.

Ejemplos

En los siguientes ejemplos de código, se usa el conector BigQueryIO con lecturas de tablas directas. Para usar un trabajo de exportación, omite la llamada a withMethod.

Lee registros con formato Avro

En este ejemplo, se muestra cómo usar el conector BigQueryIO para leer registros en formato Avro.

Para leer datos de BigQuery en registros con formato Avro, usa el método read(SerializableFunction). Este método toma una función definida por la aplicación que analiza objetos SchemaAndRecord y muestra un tipo de datos personalizado. El resultado del conector es un PCollection de tu tipo de datos personalizado.

El siguiente código lee un PCollection<MyData> de una tabla de BigQuery, en la que MyData es una clase definida por la aplicación.

Java

Para autenticarte en Dataflow, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para un entorno de desarrollo local.

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

El método read toma una interfaz SerializableFunction<SchemaAndRecord, T>, que define una función para convertir de los registros de Avro en una clase de datos personalizada. En el ejemplo de código anterior, el método MyData.apply implementa esta función de conversión. La función de ejemplo analiza los campos name y age del registro de Avro y muestra una instancia de MyData.

Nota: Este código falla si la función de análisis no coincide con el esquema real de la tabla de BigQuery. Específicamente, las columnas con nombre deben existir y los tipos de datos deben ser compatibles. Además, si alguna columna de la tabla es anulable, tu código debe controlar los valores nulos.

Para especificar qué tabla de BigQuery leer, llama al método from, como se muestra en el ejemplo anterior. Para obtener más información, consulta Nombres de tablas en la documentación del conector de E/S de BigQuery.

Lectura de objetos TableRow

En este ejemplo, se muestra cómo usar el conector BigQueryIO para leer objetos TableRow.

El método readTableRows lee datos de BigQuery en un PCollection de objetos TableRow. Cada TableRow es un mapa de pares clave-valor que contiene una sola fila de datos de tabla. Para especificar la tabla de BigQuery que se leerá, llama al método from.

El siguiente código lee un PCollection<TableRows> de una tabla de BigQuery.

Java

Para autenticarte en Dataflow, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para un entorno de desarrollo local.

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

En este ejemplo, también se muestra cómo acceder a los valores del diccionario TableRow. Los valores enteros se codifican como cadenas para coincidir con el formato de JSON exportado de BigQuery.

Nota: Este código falla si la función de análisis no coincide con el esquema real de la tabla de BigQuery. Específicamente, las columnas con nombre deben existir y los tipos de datos deben ser compatibles. Además, si alguna columna de la tabla es anulable, tu código debe controlar los valores nulos.

¿Qué sigue?

• E/S administrada para BigQuery
• Escribe en BigQuery