Spanner-Connector mit Spark verwenden

Console

1. Öffnen Sie in der Google Cloud Console die Dataproc-Seite Cluster erstellen.
2. Klicken Sie auf dem Tab „Sicherheit verwalten“ im Bereich „Projektzugriff“ auf „Aktiviert den Bereich ‚cloud-platform‘ für diesen Cluster“.
3. Füllen Sie die restlichen Felder für die Clustererstellung aus oder bestätigen Sie sie und klicken Sie dann auf „Erstellen“.

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. Prüfen Sie den Code und ersetzen Sie die Platzhalter [projectId], [instanceId], [databaseId] und [table] durch die Projekt-ID, die Instanz-ID, die Datenbank-ID und die Tabelle, die Sie zuvor erstellt haben. Mit der Option „enableDataBoost“ wird die Data Boost-Funktion von Spanner aktiviert, die nahezu keine Auswirkungen auf die Haupt-Spanner-Instanz hat.

   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. Code in einem Cluster ausführen
1. SSH-Verbindung zum Masterknoten des Dataproc-Clusters herstellen
   1. Rufen Sie in der Google Cloud Console die Seite Dataproc-Cluster auf und klicken Sie auf den Namen Ihres Clusters.

      

   2. Wählen Sie auf der Seite > Clusterdetails den Tab „VM-Instanzen“ aus. Klicken Sie dann rechts neben dem Namen des Cluster-Masterknotens auf SSH.

      

      
      Ein Browserfenster wird in Ihrem Basisverzeichnis auf dem Masterknoten geöffnet.

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

2. Erstellen Sie singers.scala mit dem vorinstallierten Texteditor vi, vim oder nano und fügen Sie dann den Scala-Code aus der Scala-Code-Liste ein.

   nano singers.scala
     

3. Starten Sie die spark-shell-REPL.

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

4. Führen Sie singers.scala mit dem Befehl :load singers.scala aus, um die Spanner-Singers-Tabelle zu erstellen. Die Ausgabeliste zeigt Beispiele aus der Ausgabe von „Sänger“ an.

   > :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. Prüfen Sie den Code und ersetzen Sie die Platzhalter [projectId], [instanceId], [databaseId] und [table] durch die Projekt-ID, die Instanz-ID, die Datenbank-ID und die Tabelle, die Sie zuvor erstellt haben. Mit der Option „enableDataBoost“ wird die Data Boost-Funktion von Spanner aktiviert, die nahezu keine Auswirkungen auf die Haupt-Spanner-Instanz hat.

   #!/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. Code in Ihrem Cluster ausführen
   1. SSH-Verbindung zum Masterknoten des Dataproc-Clusters herstellen
      1. Rufen Sie in der Google Cloud Console die Seite Dataproc-Cluster auf und klicken Sie auf den Namen Ihres Clusters.

         

      2. Wählen Sie auf der Seite Clusterdetails den Tab „VM-Instanzen“ aus. Klicken Sie dann rechts neben dem Namen des Cluster-Masterknotens auf SSH.

         

         
         Ein Browserfenster wird in Ihrem Basisverzeichnis auf dem primären Knoten geöffnet.

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

   2. Erstellen Sie singers.py mit dem vorinstallierten Texteditor vi, vim oder nano und fügen Sie dann den PySpark-Code aus der PySpark-Codeliste ein.

      nano singers.py
      

   3. Führen Sie singers.py mit spark-submit aus, um die Spanner-Singers-Tabelle zu erstellen.

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

      Die Ausgabe sieht so aus:

      ...
      +--------+---------+--------+---------+-----------+
      |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