MapReduce-Job mit dem BigQuery-Connector schreiben

Der Hadoop BigQuery-Connector wird standardmäßig auf allen Dataproc 1.0-1.2-Clusterknoten unter /usr/lib/hadoop/lib/ installiert. Er ist sowohl in Spark- als auch in PySpark-Umgebungen verfügbar.

Dataproc-Image-Versionen 1.3 und höher: Der BigQuery-Connector wird nicht standardmäßig in den Dataproc -Image-Versionen 1.3 und höher installiert. So verwenden Sie den Connector mit diesen Versionen:

1. Installieren Sie den BigQuery-Connector mithilfe der Initialisierungsaktion.

2. Geben Sie den BigQuery-Connector im Parameter jars an, wenn Sie einen Job senden:

   --jars=gs://hadoop-lib/bigquery/bigquery-connector-hadoop2-latest.jar

3. Fügen Sie die BigQuery-Connector-Klassen zur jar-with-dependencies der Anwendung hinzu.

Konflikte vermeiden: Wenn Ihre Anwendung eine andere Connector-Version als die in Ihrem Dataproc-Cluster bereitgestellte Connector-Version verwendet, müssen Sie entweder:

1. Erstellen Sie einen neuen Cluster unter Verwendung einer Initialisierungsaktion, mit der die von Ihrer Anwendung verwendete Connector-Version installiert wird.

2. Die Connector-Klassen und -Abhängigkeiten für die von Ihnen verwendete Version in die JAR-Datei Ihrer Anwendung einfügen und verschieben. Auf diese Weise vermeiden Sie Konflikte zwischen Ihrer Connector-Version und der auf Ihrem Dataproc-Cluster bereitgestellten Connector-Version. Weitere Informationen dazu finden Sie im Beispiel für die Verschiebung von Abhängigkeiten in Maven.

GsonBigQueryInputFormat-Klasse

BigQueryInputFormat wurde in GsonBigQueryInputFormat umbenannt, um das auf Gson} basierende Format hervorzuheben.

GsonBigQueryInputFormat stellt Hadoop BigQuery-Objekte in einem JsonObject-Format über die folgenden primären Vorgänge bereit:

• Verwenden einer benutzerdefinierten Abfrage zum Auswählen von BigQuery-Objekten.
• Gleichmäßiges Aufteilen der Ergebnisse der Abfrage auf Hadoop-Knoten.
• Parsen der Splitergebnisse in Java-Objekte zur Übergabe an den Mapper. Die Hadoop-Mapper-Klasse erhält eine JsonObject-Darstellung für jedes ausgewählte BigQuery-Objekt.

Die BigQueryInputFormat-Klasse bietet Zugriff auf BigQuery-Datensätze über eine Erweiterung der Hadoop InputFormat-Klasse. So verwenden Sie die BigQueryInputFormat-Klasse:

1. Zeilen müssen dem Hauptjob von Hadoop hinzugefügt werden, um Parameter in der Hadoop-Konfiguration festzulegen.

2. Die Klasse InputFormat muss auf GsonBigQueryInputFormat festgelegt sein.

In den folgenden Abschnitten erfahren Sie, wie Sie diese Anforderungen erfüllen.

Eingabeparameter

QualifiedInputTableId

Die BigQuery-Tabelle, aus der gelesen werden soll, im Format: optional-projectId:datasetId.tableId
Beispiel: publicdata:samples.shakespeare

projectId

Die Bezeichnung des BigQuery-Projekts, unter dem alle Eingabevorgänge stattfinden.
Beispiel: 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);

Hinweise:

• job bezieht sich auf org.apache.hadoop.mapreduce.Job, den auszuführenden Hadoop-Job.
• conf bezieht sich auf org.apache.hadoop.Configuration für den Hadoop-Job.

Mapper

Die Klasse GsonBigQueryInputFormat liest aus BigQuery. Die BigQuery-Objekte werden nacheinander als Eingabe an die Hadoop-Funktion Mapper übergeben. Die Eingaben erfolgen in Form von LongWritable und JsonObject. LongWritable verfolgt die Datensatznummer, JsonObject enthält den BigQuery-Datensatz im JSON-Format. Der Mapper akzeptiert das LongWritable- und JsonObject-Paar als Eingabe. Hier ist ein Snippet aus dem Mapper für einen Beispiel-Wordcount-Job.

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

IndirectBigQueryOutputFormat-Klasse

IndirectBigQueryOutputFormat bietet Hadoop die Möglichkeit, JsonObject-Werte direkt in eine BigQuery-Tabelle zu schreiben. Diese Klasse bietet über eine Erweiterung der Hadoop-Klasse OutputFormat Zugriff auf BigQuery-Datensätze. Zur korrekten Verwendung müssen mehrere Parameter in der Hadoop-Konfiguration festgelegt werden und die OutputFormat-Klasse muss auf IndirectBigQueryOutputFormat festgelegt sein. Im Folgenden finden Sie ein Beispiel für die notwendigen Parameter und Codezeilen zur korrekten Verwendung von IndirectBigQueryOutputFormat.

IndirectBigQueryOutputFormat speichert zuerst alle Daten in eine temporäre Cloud Storage-Tabelle zwischen und kopiert dann bei commitJob alle Daten aus dem Cloud Storage in einem Vorgang in BigQuery. Ihre Verwendung wird für große Jobs empfohlen, da nur ein einziger BigQuery-"Ladejob" pro Hadoop/Spark-Job erforderlich ist. Im Gegensatz dazu führt BigQueryOutputFormat für jeden Hadoop/Spark-Job einen BigQuery-Job aus.

Ausgabeparameter

projectId

Die Bezeichnung des BigQuery-Projekts, unter dem alle Ausgabevorgänge stattfinden.
Beispiel: "my-first-cloud-project"

QualifiedOutputTableId

Das BigQuery-Dataset, in das die endgültigen Jobergebnisse geschrieben werden sollen, im Format optional-projectId:datasetId.tableId. Die Dataset-ID sollte im Projekt bereits vorhanden sein. Das Dataset outputDatasetId_hadoop_temporary wird in BigQuery für die Aufnahme temporärer Ergebnisse erstellt. Überprüfen Sie, dass es nicht mit vorhandenen Datasets in Konflikt steht.
Beispiele:
test_output_dataset.wordcount_output
my-first-cloud-project:test_output_dataset.wordcount_output

outputTableFieldSchema

Ein Schema, das das Schema für die BigQuery-Ausgabetabelle definiert

GcsOutputPath

Der Ausgabepfad zum Speichern temporärer Cloud Storage-Daten (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");
);

Reducer

Die IndirectBigQueryOutputFormat-Klasse schreibt in BigQuery. Sie erhält einen Schlüssel und einen JsonObject-Wert als Eingabe, schreibt aber nur den JsonObject-Wert in BigQuery (der Schlüssel wird ignoriert). Das JsonObject sollte einen BigQuery-Datensatz im Json-Format enthalten. Der Reducer sollte ein Paar aus einem Schlüssel eines beliebigen Typs (in unserem Beispiel-WordCount-Job wird NullWritable verwendet) und einem JsonObject-Wert ausgeben. Der Reducer für den Beispiel-WordCount-Job wird unten angezeigt.

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

Bereinigen

Bereinigen Sie nach Abschluss des Jobs die Cloud Storage-Exportpfade.

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

Die Wortzahl wird in der BigQuery-Ausgabetabelle in der Google Cloud Console angezeigt.

Vollständiger Code für einen Beispiel-WordCount-Job

Der folgende Code ist ein Beispiel für einen einfachen WordCount-Job, mit dem Wortzahlen aus Objekten in BigQuery aggregiert werden.

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/.
  }
}

Java-Version

Der BigQuery-Connector erfordert Java 8.

Informationen zu Apache Maven Dependency

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

Weitere Informationen finden Sie in den Versionshinweisen zu BigQuery-Connector und in der Javadoc-Referenz.