Se usó la API de Cloud Translation para traducir esta página.

Escribe un trabajo de MapReduce con el conector de BigQuery

El conector de BigQuery de Hadoop se instala de forma predeterminada en todos los nodos del clúster de Dataproc 1.0-1.2 en /usr/lib/hadoop/lib/. Está disponible en entornos de Spark y PySpark.

Versiones 1.5 o posteriores de las imágenes de Dataproc: la versión de BigQuery El conector no está instalado de forma predeterminada en Dataproc de imágenes, versiones posteriores y 1.5. Para usarla con estas versiones, haz lo siguiente:

Instala el conector de BigQuery con la acción de inicialización.
Especifica el conector de BigQuery en el parámetro jars cuando envías un trabajo:
```
--jars=gs://hadoop-lib/bigquery/bigquery-connector-hadoop3-latest.jar
```
Incluye las clases de conector de BigQuery en el jar-with-dependencies de la aplicación.

Para evitar conflictos: Si la aplicación utiliza una versión de conector distinta de la versión de conector implementada en el clúster de Dataproc, debes hacer lo siguiente:

Crea un clúster nuevo con una acción de inicialización que instale la versión del conector que usa tu aplicación, o
Incluye y reubica las clases y dependencias del conector de la versión que usas en el archivo JAR de tu aplicación para evitar conflictos entre la versión del conector y la versión del conector implementada en tu clúster de Dataproc (consulta este ejemplo de reubicación de dependencias en Maven).

Clase GsonBigQueryInputFormat

BigQueryInputFormat cambió su nombre a GsonBigQueryInputFormat para enfatizar su formato basado en Gson}.

GsonBigQueryInputFormat proporciona a Hadoop los objetos de BigQuery en un formato JsonObject mediante las siguientes operaciones principales:

Realiza una consulta especificada por el usuario para seleccionar objetos de BigQuery.
Divide los resultados de la consulta de manera uniforme entre los nodos de Hadoop.
Analiza las divisiones en objetos Java para pasarlos al asignador. La clase Mapper de Hadoop recibe una representación JsonObject de cada objeto de BigQuery seleccionado.

La clase BigQueryInputFormat proporciona acceso a los registros de BigQuery a través de una extensión de la clase InputFormat de Hadoop. Para usar la clase BigQueryInputFormat, haz lo siguiente:

Se deben agregar líneas al trabajo principal de Hadoop para definir parámetros en la configuración de Hadoop.
La clase InputFormat debe establecerse en GsonBigQueryInputFormat.

En las siguientes secciones, se muestra cómo cumplir con estos requisitos.

Parámetros de entrada

QualifiedInputTableId: La tabla de BigQuery desde la que se realiza la lectura: optional-projectId:datasetId.tableId
Ejemplo: publicdata:samples.shakespeare
ID de proyecto: El ID de proyecto de BigQuery en el que se producen todas las operaciones de entrada.
Ejemplo: my-first-cloud-project

// Set the job-level projectId.
conf.set(BigQueryConfiguration.PROJECT_ID_KEY, projectId);

// Configure input parameters.
BigQueryConfiguration.configureBigQueryInput(conf, inputQualifiedTableId);

// Set InputFormat.
job.setInputFormatClass(GsonBigQueryInputFormat.class);

Notas:

job se refiere a org.apache.hadoop.mapreduce.Job, el trabajo de Hadoop que se ejecutará.
conf se refiere a org.apache.hadoop.Configuration para el trabajo de Hadoop.

Asignador

La clase GsonBigQueryInputFormat lee de BigQuery y pasa Objetos de BigQuery uno a la vez como entrada al Mapper de Hadoop . Las entradas tienen la forma de un par que incluye lo siguiente:

LongWritable, el número de registro
JsonObject, el registro de BigQuery con formato JSON

Mapper acepta LongWritable y JsonObject pair como entrada.

Este es un fragmento de Mapper para un trabajo de muestra WordCount.

  // private static final LongWritable ONE = new LongWritable(1);
  // The configuration key used to specify the BigQuery field name
  // ("column name").
  public static final String WORDCOUNT_WORD_FIELDNAME_KEY =
      "mapred.bq.samples.wordcount.word.key";

  // Default value for the configuration entry specified by
  // WORDCOUNT_WORD_FIELDNAME_KEY. Examples: 'word' in
  // publicdata:samples.shakespeare or 'repository_name'
  // in publicdata:samples.github_timeline.
  public static final String WORDCOUNT_WORD_FIELDNAME_VALUE_DEFAULT = "word";

  /**
   * The mapper function for WordCount.
   */
  public static class Map
      extends Mapper <LongWritable, JsonObject, Text, LongWritable> {
    private static final LongWritable ONE = new LongWritable(1);
    private Text word = new Text();
    private String wordKey;

    @Override
    public void setup(Context context)
        throws IOException, InterruptedException {
      // Find the runtime-configured key for the field name we're looking for
      // in the map task.
      Configuration conf = context.getConfiguration();
      wordKey = conf.get(WORDCOUNT_WORD_FIELDNAME_KEY,
          WORDCOUNT_WORD_FIELDNAME_VALUE_DEFAULT);
    }

    @Override
    public void map(LongWritable key, JsonObject value, Context context)
        throws IOException, InterruptedException {
      JsonElement countElement = value.get(wordKey);
      if (countElement != null) {
        String wordInRecord = countElement.getAsString();
        word.set(wordInRecord);
        // Write out the key, value pair (write out a value of 1, which will be
        // added to the total count for this word in the Reducer).
        context.write(word, ONE);
      }
    }
  }

Clase IndirectBigQueryOutputFormat

IndirectBigQueryOutputFormat proporciona a Hadoop la capacidad de escribir valores JsonObject directamente en una tabla de BigQuery. Esta clase proporciona acceso a registros de BigQuery a través de una extensión de Hadoop OutputFormat clase. Para usarla correctamente, se deben establecer varios parámetros en la configuración de Hadoop y la clase OutputFormat debe configurarse en IndirectBigQueryOutputFormat. A continuación, se muestra un ejemplo de los parámetros para configurar y las líneas de código necesarias para usar correctamente IndirectBigQueryOutputFormat.

IndirectBigQueryOutputFormat primero almacena en búfer todos los datos en una tabla temporal de Cloud Storage y, luego, en commitJob, copia todos los datos de Cloud Storage en BigQuery en una operación. Se recomienda su uso para trabajos grandes, ya que solo requiere un trabajo de "carga" de BigQuery por trabajo de Hadoop/Spark, en comparación con BigQueryOutputFormat, que realiza un trabajo de BigQuery por cada tarea de Hadoop/Spark.

Parámetros de resultado

ID de proyecto: El ID de proyecto de BigQuery en el que se producen todas las operaciones de salida.
Ejemplo: “my-first-cloud-project”
QualifiedOutputTableId: El conjunto de datos de BigQuery en el que se escriben los resultados finales del trabajo, con el formato optional-projectId:datasetId.tableId. El datasetId ya debe estar presente en el proyecto. outputDatasetId_hadoop_temporary dataset se creará el conjunto de datos_hadoop_temporal en BigQuery para obtener resultados temporales. Asegúrate de que esto no entre en conflicto con un conjunto de datos existente.
Ejemplos:
test_output_dataset.wordcount_output
my-first-cloud-project:test_output_dataset.wordcount_output
outputTableFieldSchema: Un esquema que define el esquema para la tabla de BigQuery de salida
GcsOutputPath: La ruta de salida para almacenar los datos temporales de Cloud Storage (gs://bucket/dir/)

    // Define the schema we will be using for the output BigQuery table.
    List<TableFieldSchema> outputTableFieldSchema = new ArrayList<TableFieldSchema>();
    outputTableFieldSchema.add(new TableFieldSchema().setName("Word").setType("STRING"));
    outputTableFieldSchema.add(new TableFieldSchema().setName("Count").setType("INTEGER"));
    TableSchema outputSchema = new TableSchema().setFields(outputTableFieldSchema);

    // Create the job and get its configuration.
    Job job = new Job(parser.getConfiguration(), "wordcount");
    Configuration conf = job.getConfiguration();

    // Set the job-level projectId.
    conf.set(BigQueryConfiguration.PROJECT_ID_KEY, projectId);

    // Configure input.
    BigQueryConfiguration.configureBigQueryInput(conf, inputQualifiedTableId);

    // Configure output.
    BigQueryOutputConfiguration.configure(
        conf,
        outputQualifiedTableId,
        outputSchema,
        outputGcsPath,
        BigQueryFileFormat.NEWLINE_DELIMITED_JSON,
        TextOutputFormat.class);

    // (Optional) Configure the KMS key used to encrypt the output table.
    BigQueryOutputConfiguration.setKmsKeyName(
        conf,
        "projects/myproject/locations/us-west1/keyRings/r1/cryptoKeys/k1");
);

Reductor

La clase IndirectBigQueryOutputFormat escribe en BigQuery. Toma una clave y un valor JsonObject como entrada y escribe solo el valor JsonObject en BigQuery (la clave se ignora). La JsonObject debe contener un registro de BigQuery con formato Json. El reductor debería generar una clave de cualquier tipo (NullWritable se usa en nuestro trabajo WordCount de muestra) y el par de valores JsonObject. A continuación, se muestra el reductor para el trabajo de WordCount de muestra.

  /**
   * Reducer function for WordCount.
   */
  public static class Reduce
      extends Reducer<Text, LongWritable, JsonObject, NullWritable> {

    @Override
    public void reduce(Text key, Iterable<LongWritable> values, Context context)
        throws IOException, InterruptedException {
      // Add up the values to get a total number of occurrences of our word.
      long count = 0;
      for (LongWritable val : values) {
        count = count + val.get();
      }

      JsonObject jsonObject = new JsonObject();
      jsonObject.addProperty("Word", key.toString());
      jsonObject.addProperty("Count", count);
      // Key does not matter.
      context.write(jsonObject, NullWritable.get());
    }
  }

Limpia

Una vez que se complete el trabajo, limpia las rutas de exportación de Cloud Storage.

job.waitForCompletion(true);
GsonBigQueryInputFormat.cleanupJob(job.getConfiguration(), job.getJobID());

Puedes ver los recuentos de palabras en la tabla de salida de BigQuery en la consola de Google Cloud.

Código completo para el trabajo de WordCount de muestra

El siguiente código es un ejemplo de un trabajo de WordCount simple que agrega recuentos de palabras de objetos en BigQuery.

package com.google.cloud.hadoop.io.bigquery.samples;

import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableSchema;
import com.google.cloud.hadoop.io.bigquery.BigQueryConfiguration;
import com.google.cloud.hadoop.io.bigquery.BigQueryFileFormat;
import com.google.cloud.hadoop.io.bigquery.GsonBigQueryInputFormat;
import com.google.cloud.hadoop.io.bigquery.output.BigQueryOutputConfiguration;
import com.google.cloud.hadoop.io.bigquery.output.IndirectBigQueryOutputFormat;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * Sample program to run the Hadoop Wordcount example over tables in BigQuery.
 */
public class WordCount {

 // The configuration key used to specify the BigQuery field name
  // ("column name").
  public static final String WORDCOUNT_WORD_FIELDNAME_KEY =
      "mapred.bq.samples.wordcount.word.key";

  // Default value for the configuration entry specified by
  // WORDCOUNT_WORD_FIELDNAME_KEY. Examples: 'word' in
  // publicdata:samples.shakespeare or 'repository_name'
  // in publicdata:samples.github_timeline.
  public static final String WORDCOUNT_WORD_FIELDNAME_VALUE_DEFAULT = "word";

  // Guava might not be available, so define a null / empty helper:
  private static boolean isStringNullOrEmpty(String toTest) {
    return toTest == null || "".equals(toTest);
  }

  /**
   * The mapper function for WordCount. For input, it consumes a LongWritable
   * and JsonObject as the key and value. These correspond to a row identifier
   * and Json representation of the row's values/columns.
   * For output, it produces Text and a LongWritable as the key and value.
   * These correspond to the word and a count for the number of times it has
   * occurred.
   */

  public static class Map
      extends Mapper <LongWritable, JsonObject, Text, LongWritable> {
    private static final LongWritable ONE = new LongWritable(1);
    private Text word = new Text();
    private String wordKey;

    @Override
    public void setup(Context context)
        throws IOException, InterruptedException {
      // Find the runtime-configured key for the field name we're looking for in
      // the map task.
      Configuration conf = context.getConfiguration();
      wordKey = conf.get(WORDCOUNT_WORD_FIELDNAME_KEY, WORDCOUNT_WORD_FIELDNAME_VALUE_DEFAULT);
    }

    @Override
    public void map(LongWritable key, JsonObject value, Context context)
        throws IOException, InterruptedException {
      JsonElement countElement = value.get(wordKey);
      if (countElement != null) {
        String wordInRecord = countElement.getAsString();
        word.set(wordInRecord);
        // Write out the key, value pair (write out a value of 1, which will be
        // added to the total count for this word in the Reducer).
        context.write(word, ONE);
      }
    }
  }

  /**
   * Reducer function for WordCount. For input, it consumes the Text and
   * LongWritable that the mapper produced. For output, it produces a JsonObject
   * and NullWritable. The JsonObject represents the data that will be
   * loaded into BigQuery.
   */
  public static class Reduce
      extends Reducer<Text, LongWritable, JsonObject, NullWritable> {

    @Override
    public void reduce(Text key, Iterable<LongWritable> values, Context context)
        throws IOException, InterruptedException {
      // Add up the values to get a total number of occurrences of our word.
      long count = 0;
      for (LongWritable val : values) {
        count = count + val.get();
      }

      JsonObject jsonObject = new JsonObject();
      jsonObject.addProperty("Word", key.toString());
      jsonObject.addProperty("Count", count);
      // Key does not matter.
      context.write(jsonObject, NullWritable.get());
    }
  }

  /**
   * Configures and runs the main Hadoop job. Takes a String[] of 5 parameters:
   * [ProjectId] [QualifiedInputTableId] [InputTableFieldName]
   * [QualifiedOutputTableId] [GcsOutputPath]
   *
   * ProjectId - Project under which to issue the BigQuery
   * operations. Also serves as the default project for table IDs that don't
   * specify a project for the table.
   *
   * QualifiedInputTableId - Input table ID of the form
   * (Optional ProjectId):[DatasetId].[TableId]
   *
   * InputTableFieldName - Name of the field to count in the
   * input table, e.g., 'word' in publicdata:samples.shakespeare or
   * 'repository_name' in publicdata:samples.github_timeline.
   *
   * QualifiedOutputTableId - Input table ID of the form
   * (Optional ProjectId):[DatasetId].[TableId]
   *
   * GcsOutputPath - The output path to store temporary
   * Cloud Storage data, e.g., gs://bucket/dir/
   *
   * @param args a String[] containing ProjectId, QualifiedInputTableId,
   *     InputTableFieldName, QualifiedOutputTableId, and GcsOutputPath.
   * @throws IOException on IO Error.
   * @throws InterruptedException on Interrupt.
   * @throws ClassNotFoundException if not all classes are present.
   */
  public static void main(String[] args)
      throws IOException, InterruptedException, ClassNotFoundException {

    // GenericOptionsParser is a utility to parse command line arguments
    // generic to the Hadoop framework. This example doesn't cover the specifics,
    // but recognizes several standard command line arguments, enabling
    // applications to easily specify a NameNode, a ResourceManager, additional
    // configuration resources, etc.
    GenericOptionsParser parser = new GenericOptionsParser(args);
    args = parser.getRemainingArgs();

    // Make sure we have the right parameters.
    if (args.length != 5) {
      System.out.println(
          "Usage: hadoop jar bigquery_wordcount.jar [ProjectId] [QualifiedInputTableId] "
              + "[InputTableFieldName] [QualifiedOutputTableId] [GcsOutputPath]\n"
              + "    ProjectId - Project under which to issue the BigQuery operations. Also serves "
              + "as the default project for table IDs that don't explicitly specify a project for "
              + "the table.\n"
              + "    QualifiedInputTableId - Input table ID of the form "
              + "(Optional ProjectId):[DatasetId].[TableId]\n"
              + "    InputTableFieldName - Name of the field to count in the input table, e.g., "
              + "'word' in publicdata:samples.shakespeare or 'repository_name' in "
              + "publicdata:samples.github_timeline.\n"
              + "    QualifiedOutputTableId - Input table ID of the form "
              + "(Optional ProjectId):[DatasetId].[TableId]\n"
              + "    GcsOutputPath - The output path to store temporary Cloud Storage data, e.g., "
              + "gs://bucket/dir/");
      System.exit(1);
    }

    // Get the individual parameters from the command line.
    String projectId = args[0];
    String inputQualifiedTableId = args[1];
    String inputTableFieldId = args[2];
    String outputQualifiedTableId = args[3];
    String outputGcsPath = args[4];

   // Define the schema we will be using for the output BigQuery table.
    List<TableFieldSchema> outputTableFieldSchema = new ArrayList<TableFieldSchema>();
    outputTableFieldSchema.add(new TableFieldSchema().setName("Word").setType("STRING"));
    outputTableFieldSchema.add(new TableFieldSchema().setName("Count").setType("INTEGER"));
    TableSchema outputSchema = new TableSchema().setFields(outputTableFieldSchema);

    // Create the job and get its configuration.
    Job job = new Job(parser.getConfiguration(), "wordcount");
    Configuration conf = job.getConfiguration();

    // Set the job-level projectId.
    conf.set(BigQueryConfiguration.PROJECT_ID_KEY, projectId);

    // Configure input.
    BigQueryConfiguration.configureBigQueryInput(conf, inputQualifiedTableId);

    // Configure output.
    BigQueryOutputConfiguration.configure(
        conf,
        outputQualifiedTableId,
        outputSchema,
        outputGcsPath,
        BigQueryFileFormat.NEWLINE_DELIMITED_JSON,
        TextOutputFormat.class);

    // (Optional) Configure the KMS key used to encrypt the output table.
    BigQueryOutputConfiguration.setKmsKeyName(
        conf,
        "projects/myproject/locations/us-west1/keyRings/r1/cryptoKeys/k1");

    conf.set(WORDCOUNT_WORD_FIELDNAME_KEY, inputTableFieldId);

    // This helps Hadoop identify the Jar which contains the mapper and reducer
    // by specifying a class in that Jar. This is required if the jar is being
    // passed on the command line to Hadoop.
    job.setJarByClass(WordCount.class);

    // Tell the job what data the mapper will output.
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(LongWritable.class);
    job.setMapperClass(Map.class);
    job.setReducerClass(Reduce.class);
    job.setInputFormatClass(GsonBigQueryInputFormat.class);

    // Instead of using BigQueryOutputFormat, we use the newer
    // IndirectBigQueryOutputFormat, which works by first buffering all the data
    // into a Cloud Storage temporary file, and then on commitJob, copies all data from
    // Cloud Storage into BigQuery in one operation. Its use is recommended for large jobs
    // since it only requires one BigQuery "load" job per Hadoop/Spark job, as
    // compared to BigQueryOutputFormat, which performs one BigQuery job for each
    // Hadoop/Spark task.
    job.setOutputFormatClass(IndirectBigQueryOutputFormat.class);

    job.waitForCompletion(true);

    // After the job completes, clean up the Cloud Storage export paths.
    GsonBigQueryInputFormat.cleanupJob(job.getConfiguration(), job.getJobID());

    // You can view word counts in the BigQuery output table at
    // https://console.cloud.google.com/.
  }
}

Versión de Java

El conector de BigQuery requiere Java 8.

Información sobre las dependencias de Apache Maven

<dependency>
    <groupId>com.google.cloud.bigdataoss</groupId>
    <artifactId>bigquery-connector</artifactId>
    <version>insert "hadoopX-X.X.X" connector version number here</version>
</dependency>

Para obtener información detallada, consulta las notas de la versión del conector de BigQuery y la referencia de Javadoc.