Use o conector do BigQuery com o Spark

O conector do BigQuery pode ser usado com o Apache Spark para ler e gravar dados no BigQuery e a partir dele. Veja nesta página um exemplo de código que usa o conector do BigQuery com o Spark. Consulte os Guias de início rápido do Cloud Dataproc para mais instruções sobre como criar um cluster.

Como gravar e ler dados do BigQuery

Este exemplo lê dados do BigQuery e os transfere ao Spark para realizar uma contagem de palavras usando SparkContext.newAPIHadoopRDD. Consulte a documentação do Spark para mais informações. Ele retorna os dados ao BigQuery usando PairRDDFunctions.saveAsNewAPIHadoopDataset.

Antes de testar este exemplo, crie um conjunto de dados chamado "wordcount_dataset" ou altere o conjunto de dados de saída no código para um conjunto de dados do BigQuery no projeto do Google Cloud Platform. Veja abaixo o comando bq para criar o conjunto de dados:

bq mk wordcount_dataset

Scala

Examinar o código

IndirectBigQueryOutputFormat permite ao Hadoop gravar valores JsonObject diretamente em uma tabela do BigQuery. Essa classe fornece acesso a registros do BigQuery por meio de uma extensão da classe OutputFormat do Hadoop. Para usá-la corretamente, diversos parâmetros precisam ser ajustados na configuração do Hadoop, e a classe OutputFormat precisa ser configurada como IndirectBigQueryOutputFormat. Veja abaixo um exemplo dos parâmetros a serem configurados e as linhas de código necessárias para usar IndirectBigQueryOutputFormat corretamente.
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.JsonObject
import org.apache.hadoop.io.LongWritable
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat

// Assumes you have a spark context (sc) -- running from spark-shell REPL.
// Marked as transient since configuration is not Serializable. This should
// only be necessary in spark-shell REPL.
@transient
val conf = sc.hadoopConfiguration

// Input parameters.
val fullyQualifiedInputTableId = "publicdata:samples.shakespeare"
val projectId = conf.get("fs.gs.project.id")
val bucket = conf.get("fs.gs.system.bucket")

// Input configuration.
conf.set(BigQueryConfiguration.PROJECT_ID_KEY, projectId)
conf.set(BigQueryConfiguration.GCS_BUCKET_KEY, bucket)
BigQueryConfiguration.configureBigQueryInput(conf, fullyQualifiedInputTableId)

// Output parameters.
val outputTableId = projectId + ":wordcount_dataset.wordcount_output"
// Temp output bucket that is deleted upon completion of job.
val outputGcsPath = ("gs://" + bucket + "/hadoop/tmp/bigquery/wordcountoutput")

// Output configuration.
// Let BigQuery auto-detect output schema (set to null below).
BigQueryOutputConfiguration.configureWithAutoSchema(
    conf,
    outputTableId,
    outputGcsPath,
    BigQueryFileFormat.NEWLINE_DELIMITED_JSON,
    classOf[TextOutputFormat[_,_]])

// (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("mapreduce.job.outputformat.class",
         classOf[IndirectBigQueryOutputFormat[_,_]].getName)

// Truncate the table before writing output to allow multiple runs.
conf.set(BigQueryConfiguration.OUTPUT_TABLE_WRITE_DISPOSITION_KEY,
         "WRITE_TRUNCATE")

// Helper to convert JsonObjects to (word, count) tuples.
def convertToTuple(record: JsonObject) : (String, Long) = {
  val word = record.get("word").getAsString.toLowerCase
  val count = record.get("word_count").getAsLong
  return (word, count)
}

// Helper to convert (word, count) tuples to JsonObjects.
def convertToJson(pair: (String, Long)) : JsonObject = {
  val word = pair._1
  val count = pair._2
  val jsonObject = new JsonObject()
  jsonObject.addProperty("word", word)
  jsonObject.addProperty("word_count", count)
  return jsonObject
}

// Load data from BigQuery.
val tableData = sc.newAPIHadoopRDD(
    conf,
    classOf[GsonBigQueryInputFormat],
    classOf[LongWritable],
    classOf[JsonObject])

// Perform word count.
val wordCounts = (tableData
    .map(entry => convertToTuple(entry._2))
    .reduceByKey(_ + _))

// Display 10 results.
wordCounts.take(10).foreach(l => println(l))

// Write data back into a new BigQuery table.
// IndirectBigQueryOutputFormat discards keys, so set key to null.
(wordCounts
    .map(pair => (null, convertToJson(pair)))
    .saveAsNewAPIHadoopDataset(conf))

Executar o código no seu cluster

  1. Incorpore o SSH ao node mestre do cluster do Cloud Dataproc.
    1. Acesse a página Clusters do Cloud Dataproc do seu projeto no Console do GCP e clique no nome do seu cluster.
    2. Na página de detalhes do cluster, selecione a guia "Instâncias de VM" e clique na seleção de SSH que aparece à direita do nome do node mestre do cluster.

      Uma janela de navegador é aberta no diretório principal do node mestre.
          Connected, host fingerprint: ssh-rsa 2048 ...
          ...
          user@clusterName-m:~$
          
  2. Crie wordcount.scala com o editor de texto vi, vim ou nano pré-instalado e cole o código Scala da lista de códigos Scala.
    nano wordcount.scala
      
  3. Inicie o REPL do spark-shell (remova o parâmetro jars se estiver usando o Cloud Dataproc 1.0-1.2).
    $ spark-shell --jars=gs://hadoop-lib/bigquery/bigquery-connector-hadoop2-latest.jar
    ...
    Using Scala version ...
    Type in expressions to have them evaluated.
    Type :help for more information.
    ...
    Spark context available as sc.
    ...
    SQL context available as sqlContext.
    scala>
    
  4. Execute wordcount.scala com o comando :load wordcount.scala para criar a tabela wordcount_output do BigQuery. A listagem de saída exibe 10 linhas a partir da saída de wordcount.
    :load wordcount.scala
    ...
    (pinnace,3)
    (bone,21)
    (lug,2)
    (vailing,2)
    (bombast,3)
    (gaping,11)
    (hem,5)
    ('non,1)
    (stinks,1)
    (forsooth,48)
    

    Para visualizar a tabela de saída, abra a página do BigQuery do seu projeto, selecione a tabela wordcount_output e clique em Visualizar.

PySpark

Examinar o código

#!/usr/bin/python
"""BigQuery I/O PySpark example."""
from __future__ import absolute_import
import json
import pprint
import subprocess
import pyspark
from pyspark.sql import SQLContext

sc = pyspark.SparkContext()

# Use the Cloud Storage bucket for temporary BigQuery export data used
# by the InputFormat. This assumes the Cloud Storage connector for
# Hadoop is configured.
bucket = sc._jsc.hadoopConfiguration().get('fs.gs.system.bucket')
project = sc._jsc.hadoopConfiguration().get('fs.gs.project.id')
input_directory = 'gs://{}/hadoop/tmp/bigquery/pyspark_input'.format(bucket)

conf = {
    # Input Parameters.
    'mapred.bq.project.id': project,
    'mapred.bq.gcs.bucket': bucket,
    'mapred.bq.temp.gcs.path': input_directory,
    'mapred.bq.input.project.id': 'publicdata',
    'mapred.bq.input.dataset.id': 'samples',
    'mapred.bq.input.table.id': 'shakespeare',
}

# Output Parameters.
output_dataset = 'wordcount_dataset'
output_table = 'wordcount_output'

# Load data in from BigQuery.
table_data = sc.newAPIHadoopRDD(
    'com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat',
    'org.apache.hadoop.io.LongWritable',
    'com.google.gson.JsonObject',
    conf=conf)

# Perform word count.
word_counts = (
    table_data
    .map(lambda record: json.loads(record[1]))
    .map(lambda x: (x['word'].lower(), int(x['word_count'])))
    .reduceByKey(lambda x, y: x + y))

# Display 10 results.
pprint.pprint(word_counts.take(10))

# Stage data formatted as newline-delimited JSON in Cloud Storage.
output_directory = 'gs://{}/hadoop/tmp/bigquery/pyspark_output'.format(bucket)
output_files = output_directory + '/part-*'

sql_context = SQLContext(sc)
(word_counts
 .toDF(['word', 'word_count'])
 .write.format('json').save(output_directory))

# Shell out to bq CLI to perform BigQuery import.
subprocess.check_call(
    'bq load --source_format NEWLINE_DELIMITED_JSON '
    '--replace '
    '--autodetect '
    '{dataset}.{table} {files}'.format(
        dataset=output_dataset, table=output_table, files=output_files
    ).split())

# Manually clean up the staging_directories, otherwise BigQuery
# files will remain indefinitely.
input_path = sc._jvm.org.apache.hadoop.fs.Path(input_directory)
input_path.getFileSystem(sc._jsc.hadoopConfiguration()).delete(input_path, True)
output_path = sc._jvm.org.apache.hadoop.fs.Path(output_directory)
output_path.getFileSystem(sc._jsc.hadoopConfiguration()).delete(
    output_path, True)

Executar o código no seu cluster

  1. Incorpore o SSH ao node mestre do cluster do Cloud Dataproc.
    1. Acesse a página Clusters do Cloud Dataproc do seu projeto no Console do GCP e clique no nome do seu cluster.
    2. Na página de detalhes do cluster, selecione a guia "Instâncias de VM" e clique na seleção de SSH que aparece à direita do nome do node mestre do cluster.

      Uma janela de navegador é aberta no diretório principal do node mestre.
          Connected, host fingerprint: ssh-rsa 2048 ...
          ...
          user@clusterName-m:~$
          
  2. Crie wordcount.py com o editor de texto vi, vim ou nano pré-instalado e cole o código PySpark da lista de códigos PySpark.
    nano wordcount.py
      
  3. Execute wordcount com o comando spark-submit para criar a tabela wordcount_output do BigQuery. A listagem de saída exibe 10 linhas a partir da saída de wordcount.
    spark-submit wordcount.py
    ...
    (pinnace,3)
    (bone,21)
    (lug,2)
    (vailing,2)
    (bombast,3)
    (gaping,11)
    (hem,5)
    ('non,1)
    (stinks,1)
    (forsooth,48)
    

    Para visualizar a tabela de saída, abra a página do BigQuery do seu projeto, selecione a tabela wordcount_output e clique em Visualizar.
Esta página foi útil? Conte sua opinião sobre:

Enviar comentários sobre…

Documentação do Cloud Dataproc
Precisa de ajuda? Acesse nossa página de suporte.