使用 Spark Spanner 連接器

本頁面說明如何建立 Managed Service for Apache Spark 叢集，並使用 Spark Spanner 連接器，透過 Apache Spark 在 Spanner 中讀寫資料。

Spanner 連接器可與 Spark 搭配使用，透過 Spanner Java 程式庫，從 Spanner 資料庫讀取資料及寫入資料。Spanner 連接器支援將 Spanner 資料表和圖表讀取至 Spark DataFrames 和 GraphFrames，以及將 DataFrame 資料寫入 Spanner 資料表。

費用

在本文件中，您會使用下列 Google Cloud計費元件：

• Managed Service for Apache Spark
• Spanner
• Cloud Storage

您可以使用 Pricing Calculator，根據預測用量估算費用。

初次使用 Google Cloud 的使用者可能符合免費試用期資格。

事前準備

1. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Roles required to select or create a project

   • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
   • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.
   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

2. Verify that billing is enabled for your Google Cloud project.

3. Enable the Spanner, Managed Service for Apache Spark, and Cloud Storage APIs, if any are not already enabled.

   Roles required to enable APIs

   To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

   Enable the APIs

4. 授予必要角色。
5. 設定 Managed Service for Apache Spark 叢集。
6. 設定 Spanner 執行個體，並建立 Singers 資料庫資料表。

授予必要角色

您必須具備特定 IAM 角色，才能執行本頁中的範例。視組織政策而定，這些角色可能已授予。如要查看角色授予，請參閱「是否需要授予角色？」一節。

如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和組織的存取權」。

為確保 Compute Engine 預設服務帳戶具備建立 Managed Service for Apache Spark 叢集所需的權限，請要求管理員將專案的下列 IAM 角色授予 Compute Engine 預設服務帳戶：

重要事項：您必須將這些角色授予 Compute Engine 預設服務帳戶，而非您的使用者帳戶。如果未將角色授予正確的主體，可能會發生權限錯誤。

• Dataproc worker (roles/dataproc.worker)
• Cloud Spanner 資料庫使用者 (roles/spanner.databaseUser)
• Cloud Spanner 資料庫讀取者 (使用 Data Boost) (roles/spanner.databaseReaderWithDataBoost)

設定 Managed Service for Apache Spark 叢集

建立 Managed Service for Apache Spark 叢集，或使用以 2.1 或更新版 Managed Service for Apache Spark 映像檔建立的現有 Managed Service for Apache Spark 叢集。如果叢集是以 2.0 或更早版本的映像檔建立，則必須在建立時將 scope 屬性設為 cloud-platform 範圍。

設定 Spanner 執行個體，並建立 Singers 資料庫資料表

建立 Spanner 執行個體，並在資料庫中加入 Singers 資料表。記下 Spanner 執行個體 ID 和資料庫 ID。

搭配使用 Spanner 連接器與 Spark

Spanner 連接器適用於 Spark 3.1+ 版本。提交工作至 Managed Service for Apache Spark 叢集時，您必須在 Cloud Storage 連接器 JAR 檔案規格中指定連接器版本。

範例：使用 Spanner 連接器提交 gcloud CLI Spark 工作。

gcloud dataproc jobs submit spark \
    --jars=gs://spark-lib/spanner/spark-3.1-spanner-CONNECTOR_VERSION.jar \
    ... [other job submission flags]
  

更改下列內容：

CONNECTOR_VERSION：Spanner 連接器版本。從 GitHub GoogleCloudDataproc/spark-spanner-connector 存放區的版本清單中，選擇 Spanner 連接器版本。

注意：如果連接器在執行階段無法使用，系統會擲回 ClassNotFoundException。

讀取 Spanner 資料表

您可以使用 Python 或 Scala，透過 Spark 資料來源 API 將 Spanner 資料表資料讀取至 Spark DataFrame。

PySpark

如要執行本節的 PySpark 程式碼範例，您可以將工作提交至 Managed Service for Apache Spark，或在叢集主要執行個體上透過 spark-submit REPL 執行工作。

Managed Service for Apache Spark 工作

1. 使用本機文字編輯器或 Cloud Shell 中預先安裝的 vi、vim 或 nano 文字編輯器，建立 singers.py 檔案。
1. 填入預留位置變數後，請將下列程式碼貼到 singers.py 檔案中。請注意，Spanner Data Boost 功能已啟用，對主要 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", "PROJECT_ID") \
     .option("instanceId", "INSTANCE_ID") \
     .option("databaseId", "DATABASE_ID") \
     .option("table", "TABLE_NAME") \
     .option("enableDataBoost", "true") \
     .load()
   singers.createOrReplaceTempView('Singers')
   
   # Read from Singers
   result = spark.sql('SELECT * FROM Singers')
   result.show()
   result.printSchema()
     

   更改下列內容：

   1. PROJECT_ID：您的 Google Cloud 專案 ID。 專案 ID 會列在 Google Cloud 控制台資訊主頁的「Project info」(專案資訊) 部分。
   2. INSTANCE_ID、DATABASE_ID 和 TABLE_NAME：請參閱「設定 Spanner 執行個體和 Singers 資料庫資料表」。
2. 儲存 singers.py 檔案。
2. 使用 Google Cloud 控制台、gcloud CLI 或 REST API，將工作提交至 Managed Service for Apache Spark。

   範例：使用 Spanner 連接器提交 gcloud CLI 工作。

   gcloud dataproc jobs submit pyspark singers.py \
       --cluster=CLUSTER_NAME \
       --region=REGION \
       --jars=gs://spark-lib/spanner/spark-3.1-spanner-CONNECTOR_VERSION.jar
         

   更改下列內容：

   1. CLUSTER_NAME：新叢集的名稱。
   2. REGION：可執行工作負載的 Compute Engine 區域。
   3. CONNECTOR_VERSION：Spanner 連接器版本。從 GitHub GoogleCloudDataproc/spark-spanner-connector 存放區的版本清單中，選擇 Spanner 連接器版本。

spark-submit 工作

1. 使用 SSH 連線至 Managed Service for Apache Spark 叢集主要節點。
   1. 在 Google Cloud 控制台中，前往 Managed Service for Apache Spark 的「Clusters 」(叢集) 頁面，然後按一下叢集名稱。
   2. 在「Cluster details」(叢集詳細資料) 頁面中，選取「VM Instances」(VM 執行個體) 分頁標籤。然後點選叢集主要執行個體節點名稱右側的 SSH。

      

      瀏覽器視窗會開啟主要節點上的主目錄。

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

2. 使用預先安裝的 vi、vim 或 nano 文字編輯器，在主要節點上建立 singers.py 檔案。
   1. 將預留位置變數填入 singers.py 檔案後，將下列程式碼貼到 singers.py 檔案中。請注意，Spanner Data Boost 功能已啟用，對主要 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", "PROJECT_ID") \
        .option("instanceId", "INSTANCE_ID") \
        .option("databaseId", "DATABASE_ID") \
        .option("table", "TABLE_NAME") \
        .option("enableDataBoost", "true") \
        .load()
      singers.createOrReplaceTempView('Singers')
      
      # Read from Singers
      result = spark.sql('SELECT * FROM Singers')
      result.show()
      result.printSchema()
        

      更改下列內容：

      1. PROJECT_ID：您的 Google Cloud 專案 ID。 專案 ID 會列在 Google Cloud 控制台資訊主頁的「Project info」(專案資訊) 部分。
      2. INSTANCE_ID、DATABASE_ID 和 TABLE_NAME：請參閱「設定 Spanner 執行個體和 Singers 資料庫資料表」。
   2. 儲存 singers.py 檔案。
3. 使用 spark-submit 執行 singers.py，建立 Spanner Singers 資料表。

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

   更改下列內容：

   1. CONNECTOR_VERSION：Spanner 連接器版本。 從 GitHub GoogleCloudDataproc/spark-spanner-connector 存放區的版本清單中，選擇 Spanner 連接器版本。

   輸出內容會如下所示：

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

Scala

如要在叢集上執行 Scala 程式碼範例，請完成下列步驟：

1. 使用 SSH 連線至 Managed Service for Apache Spark 叢集主要節點。
   1. 前往 Google Cloud 控制台的「Managed Service for Apache Spark Clusters 」頁面，然後按一下叢集名稱。
   2. 在「Cluster details」(叢集詳細資料) 頁面中，選取「VM Instances」(VM 執行個體) 分頁標籤。然後點選叢集主要執行個體節點名稱右側的 SSH。

      瀏覽器視窗會開啟主要節點上的主目錄。

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

2. 使用預先安裝的 vi、vim 或 nano 文字編輯器，在主要節點上建立 singers.scala 檔案。
   1. 將下列程式碼貼到 singers.scala 檔案中。請注意，Spanner Data Boost 功能已啟用，對主要 Spanner 執行個體幾乎沒有影響。

      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()
        }
      }
        

      更改下列內容：

      1. PROJECT_ID：您的 Google Cloud 專案 ID。 專案 ID 會列在 Google Cloud 控制台資訊主頁的「Project info」(專案資訊) 部分。
      2. INSTANCE_ID、DATABASE_ID 和 TABLE_NAME：請參閱「設定 Spanner 執行個體和 Singers 資料庫資料表」。
   2. 儲存 singers.scala 檔案。
3. 啟動 spark-shell REPL。

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

   更改下列內容：

   CONNECTOR_VERSION：Spanner 連接器版本。從 GitHub GoogleCloudDataproc/spark-spanner-connector 存放區的版本清單中，選擇 Spanner 連接器版本。

4. 使用 :load singers.scala 指令執行 singers.scala，建立 Spanner Singers 資料表。輸出清單會顯示 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)
     

讀取 Spanner 圖表

Spanner 連接器支援將圖表匯出至個別節點和邊緣 DataFrames，以及直接匯出至 GraphFrames。

下列範例會將 Spanner 匯出至 GraphFrame。這項作業會使用 Spanner 連接器 jar 檔案中的 Python SpannerGraphConnector 類別，讀取 Spanner Graph。

注意：請先填入預留位置變數，再執行範例。

from pyspark.sql import SparkSession

connector_jar = "gs://spark-lib/spanner/spark-3.1-spanner-CONNECTOR_VERSION.jar"

spark = (SparkSession.builder.appName("spanner-graphframe-graphx-example")
         .config("spark.jars.packages", "graphframes:graphframes:0.8.4-spark3.5-s_2.12")
         .config("spark.jars", connector_jar)
         .getOrCreate())
spark.sparkContext.addPyFile(connector_jar)

from spannergraph import SpannerGraphConnector

connector = (SpannerGraphConnector()
             .spark(spark)
             .project("PROJECT_ID")
             .instance("INSTANCE_ID")
             .database("DATABASE_ID")
             .graph("GRAPH_ID"))

g = connector.load_graph()
g.vertices.show()
g.edges.show()

更改下列內容：

• CONNECTOR_VERSION：Spanner 連接器版本。從 GitHub GoogleCloudDataproc/spark-spanner-connector 存放區的版本清單中，選擇 Spanner 連接器版本。
• PROJECT_ID：您的 Google Cloud 專案 ID。 專案 ID 會列在 Google Cloud 控制台資訊主頁的「Project info (專案資訊) 部分。
• INSTANCE_ID、DATABASE_ID 和 TABLE_NAME：插入執行個體、資料庫和圖表 ID。

如要匯出節點和邊緣 DataFrames，而非 GraphFrame，請改用 load_dfs：

df_vertices, df_edges, df_id_map = connector.load_dfs()

詳情請參閱 GitHub 的「匯出 Spanner 圖表」一文。

寫入 Spanner 資料表

Spanner 連接器支援使用 Spark 資料來源 API，將 Spark DataFrame 寫入 Spanner 資料表。

將 DataFrame 寫入 Spanner 資料表的範例

請先填入變數，再儲存並執行程式碼。

"""Spanner PySpark write example."""
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('Spanner Write App').getOrCreate()

columns = ['id', 'name', 'email']
data = [(1, 'John Doe', '[e-mail]'), (2, 'Jane Doe', '[e-mail]')]
df = spark.createDataFrame(data, columns)

df.write.format('cloud-spanner') \
    .option("projectId", "PROJECT_ID")
    .option("instanceId", "INSTANCE_ID")
    .option("databaseId", "DATABASE_ID")
    .option("table", "TABLE_NAME")
    .mode("append") \
    .save()

替換下列內容：

• PROJECT_ID： Google Cloud 專案 ID。專案 ID 會列在 Google Cloud 控制台資訊主頁的「Project info (專案資訊) 部分。
• INSTANCE_ID、DATABASE_ID 和 TABLE_NAME：插入執行個體、資料庫和資料表 ID。

注意：詳情請參閱 GitHub 的「Writing to Spanner tables」(寫入 Spanner 資料表)。

清除所用資源

如要避免系統持續向您的 Google Cloud 帳戶收取費用，可以停止或刪除 Managed Service for Apache Spark 叢集，並刪除 Spanner 執行個體。

後續步驟

• 請參閱 pyspark.sql.DataFrame 範例。
• 如要瞭解 Spark DataFrame 支援的語言，請參閱下列內容：
  • Scala
  • Java
  • R
• 請參閱 GitHub 上的 Spark Spanner 連接器存放區。
• 請參閱「Spark 工作調整提示」。