Usar o conector do Bigtable Spark

Com o conector Bigtable Spark, é possível ler e gravar dados no Bigtable. É possível ler dados do aplicativo Spark usando o Spark SQL e DataFrames. As seguintes operações do Bigtable são compatíveis usando o conector do Spark para Bigtable:

• Gravar dados
• Ler dados
• Criar uma tabela

Neste documento, mostramos como converter uma tabela de DataFrames do Spark SQL em uma tabela do Bigtable e, em seguida, compilar e criar um arquivo JAR para enviar um job do Spark.

Status do suporte do Spark e do Scala

O conector do Bigtable para Spark é compatível com as seguintes versões do Scala:

• 2.12
• 2.13

O conector do Bigtable para Spark é compatível com as seguintes versões do Spark:

• 2.4.8
• 3.2.x
• 3.3.x
• 3.4.x
• 3.5.x

O conector do Bigtable Spark é compatível com as seguintes versões do Dataproc:

• Cluster de versão de imagem 1.5
• Cluster de versão de imagem 2.0
• 2.1 cluster image-version
• 2.2 Cluster image-version
• Versão 1.0 do ambiente de execução do Dataproc sem servidor

Calcular custos

Se você decidir usar qualquer um dos seguintes componentes faturáveis do Google Cloud, vai receber uma cobrança pelos recursos usados:

• Bigtable (o uso do emulador do Bigtable não gera cobranças)
• Dataproc
• Cloud Storage

Os preços do Dataproc se aplicam ao uso de clusters do Dataproc no Compute Engine. Os preços do Dataproc sem servidor se aplicam a cargas de trabalho e sessões executadas no Dataproc sem servidor para Spark.

Para gerar uma estimativa de custo baseada na projeção de uso deste tutorial, use a calculadora de preços.

Antes de começar

Conclua os pré-requisitos a seguir antes de usar o conector do Bigtable para Spark.

Funções exigidas

Para receber as permissões necessárias para usar o conector do Bigtable para Spark, peça ao administrador para conceder a você os seguintes papéis do IAM no projeto:

• Administrador do Bigtable (roles/bigtable.admin)(opcional): permite ler ou gravar dados e criar uma tabela.
• Usuário do Bigtable (roles/bigtable.user): permite ler ou gravar dados, mas não criar uma tabela.

Para mais informações sobre a concessão de papéis, consulte Gerenciar o acesso a projetos, pastas e organizações.

Também é possível conseguir as permissões necessárias por meio de papéis personalizados ou de outros papéis predefinidos.

Se você estiver usando o Dataproc ou o Cloud Storage, talvez sejam necessárias outras permissões. Para mais informações, consulte as permissões do Dataproc e do Cloud Storage.

Configurar o Spark

Além de criar uma instância do Bigtable, você também precisa configurar a instância do Spark. Você pode fazer isso localmente ou selecionar uma destas opções para usar o Spark com o Dataproc:

• Cluster do Dataproc
• Dataproc sem servidor

Para mais informações sobre como escolher entre um cluster do Dataproc ou uma opção sem servidor, consulte a documentação Dataproc sem servidor para Spark comparado ao Dataproc no Compute Engine .

Baixar o arquivo JAR do conector

O código-fonte do conector do Spark para Bigtable com exemplos está disponível no repositório do GitHub do conector do Spark para Bigtable.

Dependendo da sua configuração do Spark, é possível acessar o arquivo JAR da seguinte maneira:

• Se você estiver executando o PySpark localmente, faça o download do arquivo JAR do conector no local do Cloud Storage gs://spark-lib/bigtable/spark-bigtable_SCALA_VERSION-CONNECTOR_VERSION.jar.

  Substitua SCALA_VERSION por 2.12 ou 2.13, que são as únicas versões do Scala compatíveis, e substitua CONNECTOR_VERSION pela versão do conector que você quer usar.

• Para o cluster do Dataproc ou a opção sem servidor, use o arquivo JAR mais recente como um artefato que pode ser adicionado aos aplicativos Spark em Scala ou Java. Para mais informações sobre como usar o arquivo JAR como um artefato, consulte Gerenciar dependências.

• Se você estiver enviando o job PySpark para o Dataproc, use a flag gcloud dataproc jobs submit pyspark --jars para definir o URI como o local do arquivo JAR no Cloud Storage. Por exemplo, gs://spark-lib/bigtable/spark-bigtable_SCALA_VERSION-CONNECTOR_VERSION.jar.

Determinar o tipo de computação

Para jobs somente leitura, use a computação sem servidor do Data Boost, que evita impactos nos clusters de serviço de aplicativos. O aplicativo Spark precisa usar a versão 1.1.0 ou mais recente do conector do Spark para usar o Data Boost.

Para usar o Data Boost, crie um perfil de app do Data Boost e forneça o ID dele para a spark.bigtable.app_profile.id opção do Spark ao adicionar a configuração do Bigtable ao aplicativo Spark. Se você já criou um perfil de app para seus jobs de leitura do Spark e quer continuar usando sem mudar o código do aplicativo, converta o perfil de app para um perfil do Data Boost. Para mais informações, consulte Converter um perfil de app.

Para mais informações, consulte a visão geral do Bigtable Data Boost.

Para jobs que envolvem leituras e gravações, é possível usar os nós do cluster da instância para computação especificando um perfil de app padrão com sua solicitação.

Identificar ou criar um perfil de app para usar

Se você não especificar um ID de perfil de aplicativo, o conector usará o perfil de aplicativo padrão.

Recomendamos que você use um perfil de app exclusivo para cada aplicativo que executar, incluindo o aplicativo Spark. Para mais informações sobre tipos e configurações de perfis de app, consulte a Visão geral dos perfis de app. Para instruções, consulte Criar e configurar perfis de app.

Adicionar a configuração do Bigtable ao aplicativo Spark

No aplicativo Spark, adicione as opções que permitem interagir com o Bigtable.

Opções do Spark compatíveis

Use as opções do Spark disponíveis como parte do pacote com.google.cloud.spark.bigtable.

Criar metadados de tabela no formato JSON

O formato de tabela DataFrames do Spark SQL precisa ser convertido em uma tabela do Bigtable usando uma string com formato JSON. Esse formato JSON de string torna o formato de dados compatível com o Bigtable. É possível transmitir o formato JSON no código do aplicativo usando a opção .option("catalog", catalog_json_string).

Por exemplo, considere a tabela DataFrame e a tabela Bigtable correspondente a seguir.

Neste exemplo, as colunas name e birthYear no DataFrame são agrupadas no grupo de colunas info e renomeadas como name e birth_year, respectivamente. Da mesma forma, a coluna address é armazenada no grupo de colunas location com o mesmo nome de coluna. A coluna id do DataFrame é convertida na chave de linha do Bigtable.

As chaves de linha não têm um nome de coluna dedicado no Bigtable. Neste exemplo, id_rowkey é usado apenas para indicar ao conector que essa é a coluna de chave de linha. Você pode usar qualquer nome para a coluna de chave de linha. Não se esqueça de usar o mesmo nome ao declarar o campo "rowkey":"column_name" no formato JSON.

O formato JSON do catálogo é o seguinte:

    """
    {
      "table": {"name": "t1"},
      "rowkey": "id_rowkey",
      "columns": {
        "id": {"cf": "rowkey", "col": "id_rowkey", "type": "string"},
        "name": {"cf": "info", "col": "name", "type": "string"},
        "birthYear": {"cf": "info", "col": "birth_year", "type": "long"},
        "address": {"cf": "location", "col": "address", "type": "string"}
      }
    }
    """

As chaves e os valores usados no formato JSON são os seguintes:

Tipos de dados compatíveis

O conector é compatível com o uso de tipos string, long e binary (matriz de bytes) no catálogo. Até que a compatibilidade com outros tipos, como int e float, seja adicionada, é possível converter manualmente esses tipos de dados em matrizes de bytes (BinaryType do Spark SQL) antes de usar o conector para gravar no Bigtable.

Além disso, é possível usar o Avro para serializar tipos complexos, como ArrayType. Para mais informações, consulte Serializar tipos de dados complexos usando o Apache Avro.

Gravar no Bigtable

Use a função .write() e as opções compatíveis para gravar seus dados no Bigtable.

  String catalog = "{" +
        "\"table\":{\"name\":\"" + tableName + "\"," +
        "\"tableCoder\":\"PrimitiveType\"}," +
        "\"rowkey\":\"wordCol\"," +
        "\"columns\":{" +
        "\"word\":{\"cf\":\"rowkey\", \"col\":\"wordCol\", \"type\":\"string\"}," +
        "\"count\":{\"cf\":\"example_family\", \"col\":\"countCol\", \"type\":\"long\"}" +
        "}}".replaceAll("\\s+", "");

…

  private static void writeDataframeToBigtable(Dataset<Row> dataframe, String catalog,
        String createNewTable) {
      dataframe
          .write()
          .format("bigtable")
          .option("catalog", catalog)
          .option("spark.bigtable.project.id", projectId)
          .option("spark.bigtable.instance.id", instanceId)
          .option("spark.bigtable.create.new.table", createNewTable)
          .save();
    }

  catalog = ''.join(("""{
        "table":{"name":" """ + bigtable_table_name + """
        ", "tableCoder":"PrimitiveType"},
        "rowkey":"wordCol",
        "columns":{
          "word":{"cf":"rowkey", "col":"wordCol", "type":"string"},
          "count":{"cf":"example_family", "col":"countCol", "type":"long"}
        }
        }""").split())
  …

  input_data = spark.createDataFrame(data)
  print('Created the DataFrame:')
  input_data.show()

  input_data.write \
        .format('bigtable') \
        .options(catalog=catalog) \
        .option('spark.bigtable.project.id', bigtable_project_id) \
        .option('spark.bigtable.instance.id', bigtable_instance_id) \
        .option('spark.bigtable.create.new.table', create_new_table) \
        .save()
  print('DataFrame was written to Bigtable.')

  …

Ler do Bigtable

Use a função .read() para verificar se a tabela foi importada para o Bigtable.

  …
  private static Dataset<Row> readDataframeFromBigtable(String catalog) {
      Dataset<Row> dataframe = spark
          .read()
          .format("bigtable")
          .option("catalog", catalog)
          .option("spark.bigtable.project.id", projectId)
          .option("spark.bigtable.instance.id", instanceId)
          .load();
      return dataframe;
    }

  …

  records = spark.read \
        .format('bigtable') \
        .option('spark.bigtable.project.id', bigtable_project_id) \
        .option('spark.bigtable.instance.id', bigtable_instance_id) \
        .options(catalog=catalog) \
        .load()

  print('Reading the DataFrame from Bigtable:')
  records.show()

Compilar o projeto

Gere o arquivo JAR usado para executar um job em um cluster do Dataproc, no Dataproc sem servidor ou em uma instância local do Spark. É possível compilar o arquivo JAR localmente e usá-lo para enviar um job. O caminho para o JAR compilado é definido como a variável de ambiente PATH_TO_COMPILED_JAR ao enviar um job.

Essa etapa não se aplica a aplicativos PySpark.

Gerenciar dependências

O conector do Bigtable Spark é compatível com as seguintes ferramentas de gerenciamento de dependências:

• Maven
• sbt
• Gradle

Compilar o arquivo JAR

Envie um job

Envie um job do Spark usando o Dataproc, o Dataproc sem servidor ou uma instância local do Spark para iniciar o aplicativo.

Definir o ambiente de execução

Defina as seguintes variáveis de ambiente.

      #Google Cloud
      export BIGTABLE_SPARK_PROJECT_ID=PROJECT_ID
      export BIGTABLE_SPARK_INSTANCE_ID=INSTANCE_ID
      export BIGTABLE_SPARK_TABLE_NAME=TABLE_NAME
      export BIGTABLE_SPARK_DATAPROC_CLUSTER=DATAPROC_CLUSTER
      export BIGTABLE_SPARK_DATAPROC_REGION=DATAPROC_REGION
      export BIGTABLE_SPARK_DATAPROC_ZONE=DATAPROC_ZONE

      #Dataproc Serverless
      export BIGTABLE_SPARK_SUBNET=SUBNET
      export BIGTABLE_SPARK_GCS_BUCKET_NAME=GCS_BUCKET_NAME

      #Scala/Java
      export PATH_TO_COMPILED_JAR=PATH_TO_COMPILED_JAR

      #PySpark
      export GCS_PATH_TO_CONNECTOR_JAR=GCS_PATH_TO_CONNECTOR_JAR
      export PATH_TO_PYTHON_FILE=PATH_TO_PYTHON_FILE
      export LOCAL_PATH_TO_CONNECTOR_JAR=LOCAL_PATH_TO_CONNECTOR_JAR

Substitua:

• PROJECT_ID: o identificador permanente do projeto do Bigtable.
• INSTANCE_ID: o identificador permanente da instância do Bigtable.
• TABLE_NAME: identificador permanente da tabela.
• DATAPROC_CLUSTER: identificador permanente do cluster do Dataproc.
• DATAPROC_REGION: a região do Dataproc que contém um dos clusters na sua instância do Dataproc, por exemplo, northamerica-northeast2.
• DATAPROC_ZONE: a zona em que o cluster do Dataproc é executado.
• SUBNET: o caminho completo do recurso da sub-rede.
• GCS_BUCKET_NAME: o bucket do Cloud Storage para fazer upload das dependências da carga de trabalho do Spark.
• PATH_TO_COMPILED_JAR: o caminho completo ou relativo para o JAR compilado. Por exemplo, /path/to/project/root/target/<compiled_JAR_name> para Maven.
• GCS_PATH_TO_CONNECTOR_JAR: o bucket do Cloud Storage gs://spark-lib/bigtable, em que o arquivo spark-bigtable_SCALA_VERSION-CONNECTOR_VERSION.jar está localizado.
• PATH_TO_PYTHON_FILE: para aplicativos PySpark, o caminho para o arquivo Python que será usado para gravar e ler dados do Bigtable.
• LOCAL_PATH_TO_CONNECTOR_JAR: para aplicativos PySpark, o caminho para o arquivo JAR do conector do Bigtable Spark baixado.

Enviar um job do Spark

Para instâncias do Dataproc ou sua configuração local do Spark, execute um job do Spark para fazer upload de dados para o Bigtable.

Cluster do Dataproc

Use o arquivo JAR compilado e crie um job de cluster do Dataproc que lê e grava dados do e para o Bigtable.

1. Criar um cluster de Dataproc. O exemplo a seguir mostra um comando de amostra para criar um cluster do Dataproc v2.0 com Debian 10, dois nós de trabalho e configurações padrão.

   gcloud dataproc clusters create \
     $BIGTABLE_SPARK_DATAPROC_CLUSTER --region $BIGTABLE_SPARK_DATAPROC_REGION \
     --zone $BIGTABLE_SPARK_DATAPROC_ZONE \
     --master-machine-type n2-standard-4 --master-boot-disk-size 500 \
     --num-workers 2 --worker-machine-type n2-standard-4 --worker-boot-disk-size 500 \
     --image-version 2.0-debian10 --project $BIGTABLE_SPARK_PROJECT_ID
   

2. Enviar um job.

   Scala/Java

   O exemplo a seguir mostra a classe spark.bigtable.example.WordCount, que inclui a lógica para criar uma tabela de teste no DataFrame, gravar a tabela no Bigtable e contar o número de palavras na tabela.

       gcloud dataproc jobs submit spark \
       --cluster=$BIGTABLE_SPARK_DATAPROC_CLUSTER \
       --region=$BIGTABLE_SPARK_DATAPROC_REGION \
       --class=spark.bigtable.example.WordCount \
       --jar=$PATH_TO_COMPILED_JAR \
       -- \
       $BIGTABLE_SPARK_PROJECT_ID \
       $BIGTABLE_SPARK_INSTANCE_ID \
       $BIGTABLE_SPARK_TABLE_NAME \
   

   PySpark

       gcloud dataproc jobs submit pyspark \
       --cluster=$BIGTABLE_SPARK_DATAPROC_CLUSTER \
       --region=$BIGTABLE_SPARK_DATAPROC_REGION \
       --jars=$GCS_PATH_TO_CONNECTOR_JAR \
       --properties='spark.jars.packages=org.slf4j:slf4j-reload4j:1.7.36' \
       $PATH_TO_PYTHON_FILE \
       -- \
       --bigtableProjectId=$BIGTABLE_SPARK_PROJECT_ID \
       --bigtableInstanceId=$BIGTABLE_SPARK_INSTANCE_ID \
       --bigtableTableName=$BIGTABLE_SPARK_TABLE_NAME \
   

Dataproc sem servidor

Use o arquivo JAR compilado e crie um job do Dataproc que lê e grava dados do e para o Bigtable com uma instância do Dataproc Serverless.

  gcloud dataproc batches submit spark \
  --region=$BIGTABLE_SPARK_DATAPROC_REGION \
  --subnet=$BIGTABLE_SPARK_SUBNET --version=1.1 \
  --deps-bucket=gs://$BIGTABLE_SPARK_GCS_BUCKET_NAME --jar=$PATH_TO_COMPILED_JAR \
  --  \
  $BIGTABLE_SPARK_PROJECT_ID \
  $BIGTABLE_SPARK_INSTANCE_ID \
  $BIGTABLE_SPARK_TABLE_NAME

  gcloud dataproc batches submit pyspark $PATH_TO_PYTHON_FILE \
  --region=$BIGTABLE_SPARK_DATAPROC_REGION \
  --subnet=$BIGTABLE_SPARK_SUBNET --version=1.1 \
  --deps-bucket=gs://$BIGTABLE_SPARK_GCS_BUCKET_NAME \
  --jars=$GCS_PATH_TO_CONNECTOR_JAR \
  --properties='spark.jars.packages=org.slf4j:slf4j-reload4j:1.7.36' \
  -- \
  --bigtableProjectId=$BIGTABLE_SPARK_PROJECT_ID \
  --bigtableInstanceId=$BIGTABLE_SPARK_INSTANCE_ID \
  --bigtableTableName=$BIGTABLE_SPARK_TABLE_NAME

Spark local

Use o arquivo JAR baixado e crie um job do Spark que lê e grava dados do e para o Bigtable com uma instância local do Spark. Você também pode usar o emulador do Bigtable para enviar o job do Spark.

Usar o emulador do Bigtable

Se você decidir usar o emulador do Bigtable, siga estas etapas:

1. Execute o seguinte comando para iniciar o emulador:

   gcloud beta emulators bigtable start
   

   Por padrão, o emulador escolhe localhost:8086.

2. Defina a variável de ambiente BIGTABLE_EMULATOR_HOST:

   export BIGTABLE_EMULATOR_HOST=localhost:8086
   

3. Envie o job do Spark.

Para mais informações sobre como usar o emulador do Bigtable, consulte Testar usando o emulador.

Enviar um job do Spark

Use o comando spark-submit para enviar um job do Spark, mesmo que você esteja usando um emulador local do Bigtable.

  spark-submit $PATH_TO_COMPILED_JAR \
  $BIGTABLE_SPARK_PROJECT_ID \
  $BIGTABLE_SPARK_INSTANCE_ID \
  $BIGTABLE_SPARK_TABLE_NAME

  spark-submit \
  --jars=$LOCAL_PATH_TO_CONNECTOR_JAR \
  --packages=org.slf4j:slf4j-reload4j:1.7.36 \
  $PATH_TO_PYTHON_FILE \
  --bigtableProjectId=$BIGTABLE_SPARK_PROJECT_ID \
  --bigtableInstanceId=$BIGTABLE_SPARK_INSTANCE_ID \
  --bigtableTableName=$BIGTABLE_SPARK_TABLE_NAME

Verificar os dados da tabela

Execute o seguinte comando da CLI cbt para verificar se os dados foram gravados no Bigtable. A CLI cbt é um componente da Google Cloud CLI. Para mais informações, consulte a visão geral da CLI cbt.

    cbt -project=$BIGTABLE_SPARK_PROJECT_ID -instance=$BIGTABLE_SPARK_INSTANCE_ID \
    read $BIGTABLE_SPARK_TABLE_NAME

Soluções adicionais

Use o conector do Bigtable Spark para soluções específicas, como serializar tipos complexos do Spark SQL, ler linhas específicas e gerar métricas do lado do cliente.

Ler uma linha específica do DataFrame usando um filtro

Ao usar DataFrames para ler do Bigtable, é possível especificar um filtro para ler apenas linhas específicas. Filtros simples, como ==, <= e startsWith na coluna de chave de linha, são aplicados no lado do servidor para evitar uma verificação completa da tabela. Filtros em chaves de linha compostas ou filtros complexos, como o filtro LIKE na coluna de chave de linha, são aplicados no lado do cliente.

Se você estiver lendo tabelas grandes, recomendamos usar filtros simples de chave de linha para evitar uma verificação completa da tabela. A instrução de exemplo a seguir mostra como ler usando um filtro simples. Verifique se, no filtro do Spark, você usa o nome da coluna do DataFrame que é convertida na chave de linha:

    dataframe.filter("id == 'some_id'").show()
  

Ao aplicar um filtro, use o nome da coluna do DataFrame em vez do nome da coluna da tabela do Bigtable.

Serializar tipos de dados complexos usando o Apache Avro

O conector do Bigtable para Spark oferece suporte ao uso do Apache Avro para serializar tipos SQL do Spark complexos, como ArrayType, MapType ou StructType. O Apache Avro oferece serialização de dados para dados de registros, que são usados com frequência para processar e armazenar estruturas de dados complexas.

Use uma sintaxe como "avro":"avroSchema" para especificar que uma coluna no Bigtable precisa ser codificada usando Avro. Depois, use .option("avroSchema", avroSchemaString) ao ler ou gravar no Bigtable para especificar o esquema Avro correspondente a essa coluna no formato de string. É possível usar nomes de opções diferentes, por exemplo, "anotherAvroSchema" para colunas diferentes, e transmitir esquemas Avro para várias colunas.

def catalogWithAvroColumn = s"""{
                    |"table":{"name":"ExampleAvroTable"},
                    |"rowkey":"key",
                    |"columns":{
                    |"col0":{"cf":"rowkey", "col":"key", "type":"string"},
                    |"col1":{"cf":"cf1", "col":"col1", "avro":"avroSchema"}
                    |}
                    |}""".stripMargin

Usar métricas do lado do cliente

Como o conector do Bigtable para Spark é baseado no cliente do Bigtable para Java, as métricas do lado do cliente são ativadas por padrão no conector. Consulte a documentação sobre métricas do lado do cliente para mais detalhes sobre como acessar e interpretar essas métricas.

Usar o cliente Bigtable para Java com funções RDD de baixo nível

Como o conector do Bigtable para Spark é baseado no cliente do Bigtable para Java, é possível usar o cliente diretamente nos aplicativos Spark e fazer solicitações de leitura ou gravação distribuídas nas funções RDD de baixo nível, como mapPartitions e foreachPartition.

Para usar as classes do cliente do Bigtable para Java, adicione o prefixo com.google.cloud.spark.bigtable.repackaged aos nomes dos pacotes. Por exemplo, em vez de usar o nome da classe como com.google.cloud.bigtable.data.v2.BigtableDataClient, use com.google.cloud.spark.bigtable.repackaged.com.google.cloud.bigtable.data.v2.BigtableDataClient.

Para mais informações sobre o cliente Bigtable para Java, consulte o cliente Bigtable para Java.

A seguir

• Saiba como ajustar seu job do Spark no Dataproc.
• Use classes do cliente do Bigtable para Java com o conector do Bigtable para Spark.