Use the Spanner connector with Spark

PySpark

You can run the example PySpark code in this section on your cluster by submitting the job to the Dataproc service or by running the job from the spark-submit REPL on the cluster master node.

Scala

To run the example Scala code on your cluster, complete the following steps:

1. Connect to the Dataproc cluster master node using SSH.
   1. Go to the Dataproc Clusters page in the Google Cloud console, then click the name of your cluster.
   2. On the Cluster details page, select the VM Instances tab. Then click SSH to the right of the name of the cluster master node.

      

      A browser window opens at your home directory on the master node.

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

2. Create a singers.scala file on the master node using the pre-installed vi, vim, or nano text editor.
   1. Paste the following code into the singers.scala file. Note that the Spanner Data Boost feature is enabled, which has near-zero impact on the main Spanner instance.

      object singers {
        def main(): Unit = {
          /*
           * Uncomment (use the following code) 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", "PROJECT_ID")
              .option("instanceId", "INSTANCE_ID")
              .option("databaseId", "DATABASE_ID")
              .option("table", "TABLE_NAME")
              .option("enableDataBoost", true)
              .load()
              .cache())
      
          singersDF.createOrReplaceTempView("Singers")
      
          // Load the Singers table.
          val result = spark.sql("SELECT * FROM Singers")
          result.show()
          result.printSchema()
        }
      }
        

      Replace the following:

      1. PROJECT_ID: Your Google Cloud project ID. Project IDs are listed in the Project info section on the Google Cloud console Dashboard.
      2. INSTANCE_ID, DATABASE_ID, and TABLE_NAME : See Set up a Spanner instance with Singers database table.
   2. Save the singers.scala file.
3. Launch the spark-shell REPL.

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

   Replace the following:

   CONNECTOR_VERSION: Spanner connector version. Choose the Spanner connector version from the version list in the GitHub GoogleCloudDataproc/spark-spanner-connector repository.

4. Run singers.scala with the :load singers.scala command to create the Spanner Singers table. The output listing displays examplesfrom the Singers output.

   > :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)