Usar o conector do Spanner com o Spark

Console

1. No console do Google Cloud, abra o Dataproc Página Criar um cluster
2. Na página "Gerenciar segurança", clique em "Ativa o escopo da plataforma de nuvem para este cluster" na seção "Acesso ao projeto" nesta seção.
3. Preencha ou confirme os outros campos de criação de cluster e, em seguida, clique em "Criar".

Google Cloud CLI

gcloud dataproc clusters create CLUSTER_NAME --scopes https://www.googleapis.com/auth/cloud-platform
    

API

        "serviceAccountScopes": ["https://www.googleapis.com/auth/cloud-platform"],
    

Scala

1. Analise o código e substitua os marcadores de posição [projectId], [instanceId], [databaseId] e [table] por o ID do projeto, o ID da instância, o ID do banco de dados e a tabela que você criou anteriormente. A opção enableDataBoost ativa o recurso Data Boost do Spanner, que tem impacto quase zero na instância principal do Spanner.

   object singers {
     def main(): Unit = {
       /*
        * Remove comment if you are not running in spark-shell.
        *
       import org.apache.spark.sql.SparkSession
       val spark = SparkSession.builder()
         .appName("spark-spanner-demo")
         .getOrCreate()
       */
   
       // Load data in from Spanner. See
       // https://github.com/GoogleCloudDataproc/spark-spanner-connector/blob/main/README.md#properties
       // for option information.
       val singersDF =
         (spark.read.format("cloud-spanner")
           .option("projectId", "[projectId]")
           .option("instanceId", "[instanceId]")
           .option("databaseId", "[databaseId]")
           .option("enableDataBoost", true)
           .option("table", "[table]")
           .load()
           .cache())
   
       singersDF.createOrReplaceTempView("Singers")
   
       // Load the Singers table.
       val result = spark.sql("SELECT * FROM Singers")
       result.show()
       result.printSchema()
     }
   }

2. Executar o código no seu cluster
1. Use o SSH para se conectar ao nó mestre do cluster do Dataproc
   1. Acesse o Clusters do Dataproc página no console do Google Cloud e, em seguida, clique no nome do seu cluster

      

   2. Na página >Detalhes do cluster, selecione a guia "Instâncias de VM". Em seguida, clique em SSH à direita do nome do nó mestre do cluster.

      

      
      Uma janela do navegador é aberta no diretório principal do nó mestre

          Connected, host fingerprint: ssh-rsa 2048 ...
          ...
          user@clusterName-m:~$
          

2. Crie singers.scala com o editor de texto vi, vim ou nano pré-instalado e cole o código da lista de códigos Scala

   nano singers.scala
     

3. Inicie o REPL spark-shell.

   $ spark-shell --jars=gs://spark-lib/spanner/spark-3.1-spanner-CONNECTOR_VERSION.jar
   

4. Executar singers.scala com o comando :load singers.scala para criar a tabela Singers do Spanner. O resultado a listagem exibe exemplos da saída de Singers.

   > :load singers.scala
   Loading singers.scala...
   defined object singers
   > singers.main()
   ...
   +--------+---------+--------+---------+-----------+
   |SingerId|FirstName|LastName|BirthDate|LastUpdated|
   +--------+---------+--------+---------+-----------+
   |       1|     Marc|Richards|     null|       null|
   |       2| Catalina|   Smith|     null|       null|
   |       3|    Alice| Trentor|     null|       null|
   +--------+---------+--------+---------+-----------+
   
   root
    |-- SingerId: long (nullable = false)
    |-- FirstName: string (nullable = true)
    |-- LastName: string (nullable = true)
    |-- BirthDate: date (nullable = true)
    |-- LastUpdated: timestamp (nullable = true)
    

PySpark

1. Examine o código e substitua os marcadores [projectId], [instanceId], [databaseId] e [table] pelo ID do projeto, ID da instância, ID do banco de dados e tabela que você criou anteriormente. A opção enableDataBoost ativa o recurso Data Boost do Spanner, que tem quase zero na instância principal do Spanner.

   #!/usr/bin/env python
   
   """Spanner PySpark read example."""
   
   from pyspark.sql import SparkSession
   
   spark = SparkSession \
     .builder \
     .master('yarn') \
     .appName('spark-spanner-demo') \
     .getOrCreate()
   
   # Load data from Spanner.
   singers = spark.read.format('cloud-spanner') \
     .option("projectId", "[projectId]") \
     .option("instanceId", "[instanceId]") \
     .option("databaseId", "[databaseId]") \
     .option("enableDataBoost", "true") \
     .option("table", "[table]") \
     .load()
   singers.createOrReplaceTempView('Singers')
   
   # Read from Singers
   result = spark.sql('SELECT * FROM Singers')
   result.show()
   result.printSchema()

2. Executar o código no cluster
   1. Use o SSH para se conectar ao nó mestre do cluster do Dataproc
      1. Acesse o Clusters do Dataproc página no console do Google Cloud e, em seguida, clique no nome do seu cluster

         

      2. Na página Detalhes do cluster, selecione a guia "Instâncias de VM". Em seguida, clique em SSH à direita do nome do nó mestre do cluster.

         

         
         Uma janela do navegador é aberta no seu diretório principal no nó principal

             Connected, host fingerprint: ssh-rsa 2048 ...
             ...
             user@clusterName-m:~$
             

   2. Crie singers.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 singers.py
      

   3. Execute singers.py com spark-submit para criar a tabela Singers do Spanner.

      spark-submit --jars gs://spark-lib/spanner/spark-3.1-spanner-CONNECTOR_VERSION.jar singers.py
      

      A saída é:

      ...
      +--------+---------+--------+---------+-----------+
      |SingerId|FirstName|LastName|BirthDate|LastUpdated|
      +--------+---------+--------+---------+-----------+
      |       1|     Marc|Richards|     null|       null|
      |       2| Catalina|   Smith|     null|       null|
      |       3|    Alice| Trentor|     null|       null|
      +--------+---------+--------+---------+-----------+
      
      root
       |-- SingerId: long (nullable = false)
       |-- FirstName: string (nullable = true)
       |-- LastName: string (nullable = true)
       |-- BirthDate: date (nullable = true)
       |-- LastUpdated: timestamp (nullable = true)
      only showing top 20 rows