Usa el conector de Spark Spanner

En esta página, se muestra cómo crear un clúster de Managed Service para Apache Spark que usa el conector de Spark Spanner para leer y escribir datos en Spanner con Apache Spark.

El conector de Spanner funciona con Spark para leer y escribir datos en la base de datos de Spanner con la biblioteca de Java de Spanner. El conector de Spanner admite la lectura de tablas y gráficos de Spanner en DataFrames y GraphFrames de Spark, y la escritura de datos de DataFrame en tablas de Spanner.

Costos

En este documento, usarás los siguientes componentes facturables de Google Cloud:

• Managed Service for Apache Spark
• Spanner
• Cloud Storage

Para generar una estimación de costos en función del uso previsto, usa la calculadora de precios.

Es posible que los usuarios nuevos de Google Cloud cumplan con los requisitos para acceder a una prueba gratuita.

Antes de comenzar

Accede a tu cuenta de Google Cloud . Si eres nuevo en Google Cloud, crea una cuenta para evaluar el rendimiento de nuestros productos en situaciones reales. Los clientes nuevos también obtienen $300 en créditos gratuitos para ejecutar, probar y, además, implementar cargas de trabajo.

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

Verify that billing is enabled for your Google Cloud project.

Enable the Spanner, Managed Service for Apache Spark, and Cloud Storage APIs.

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

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

Verify that billing is enabled for your Google Cloud project.

Enable the Spanner, Managed Service for Apache Spark, and Cloud Storage APIs.

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

1. Otorga los roles necesarios.
2. Configura un clúster de Managed Service para Apache Spark.
3. Configura una instancia de Spanner con una tabla de base de datos de Singers.

Otorga roles necesarios

Se requieren ciertos roles de IAM para ejecutar los ejemplos de esta página. Según las políticas de la organización, es posible que estos roles ya se hayan otorgado. Para verificar las asignaciones de roles, consulta ¿Necesitas otorgar roles?.

Para obtener más información sobre cómo otorgar roles, consulta Administra el acceso a proyectos, carpetas y organizaciones.

Para garantizar que la cuenta de servicio predeterminada de Compute Engine tenga los permisos necesarios para crear un clúster de Managed Service for Apache Spark, pídele a tu administrador que otorgue los siguientes roles de IAM a la cuenta de servicio predeterminada de Compute Engine en el proyecto:

Importante: Debes otorgar estos roles a la cuenta de servicio predeterminada de Compute Engine, no a tu cuenta de usuario. Si no se otorgan los roles a la principal correcta, es posible que se produzcan errores de permisos.

• Trabajador de Dataproc (roles/dataproc.worker)
• Usuario de base de datos de Cloud Spanner (roles/spanner.databaseUser)
• Lector de base de datos de Cloud Spanner con DataBoost (roles/spanner.databaseReaderWithDataBoost)

Configura un clúster de Managed Service para Apache Spark

Crea un clúster de Managed Service para Apache Spark o usa uno existente que se haya creado con la imagen de Managed Service para Apache Spark 2.1 o posterior o, si el clúster se creó con la imagen 2.0 o anterior, debe haberse creado con la propiedad scope establecida en el alcance de cloud-platform.

Configura una instancia de Spanner con una tabla de base de datos de Singers

Crea una instancia de Spanner con una base de datos que contenga una tabla Singers. Toma nota del ID de la instancia y el ID de la base de datos de Spanner.

Usa el conector de Spanner con Spark

El conector de Spanner está disponible para las versiones 3.1+ de Spark. Especificas la versión del conector como parte de la especificación del archivo JAR del conector de Cloud Storage cuando envías un trabajo a un clúster de Managed Service para Apache Spark.

Ejemplo: Envío de un trabajo de Spark con gcloud CLI y el conector de Spanner.

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

Reemplaza lo siguiente:

CONNECTOR_VERSION: Versión del conector de Spanner. Elige la versión del conector de Spanner en la lista de versiones del repositorio de GitHub GoogleCloudDataproc/spark-spanner-connector.

Nota: Si el conector no está disponible en el tiempo de ejecución, se arroja un ClassNotFoundException.

Leer tablas de Spanner

Puedes usar Python o Scala para leer datos de tablas de Spanner en un DataFrame de Spark con la API de fuente de datos de Spark.

PySpark

Puedes ejecutar el código de ejemplo de PySpark en esta sección en tu clúster enviando el trabajo a Managed Service para Apache Spark o ejecutándolo desde el REPL de spark-submit en la instancia principal del clúster.

Trabajo de Managed Service para Apache Spark

1. Crea un archivo singers.py con un editor de texto local o en Cloud Shell con el editor de texto preinstalado vi, vim o nano.
1. Después de completar las variables de marcador de posición, pega el siguiente código en el archivo singers.py. Ten en cuenta que la función Data Boost de Spanner está habilitada, lo que tiene un impacto casi nulo en la instancia principal de 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()
     

   Reemplaza lo siguiente:

   1. PROJECT_ID: Es el ID del proyecto de Google Cloud . Los IDs del proyecto se enumeran en la sección Información del proyecto del panel de la consola de Google Cloud .
   2. INSTANCE_ID, DATABASE_ID y TABLE_NAME : Consulta Cómo configurar una instancia de Spanner con la tabla de base de datos Singers.
2. Guarda el archivo singers.py.
2. Envía el trabajo a Managed Service para Apache Spark con la consola de Google Cloud , gcloud CLI o la API de REST.

   Ejemplo: Envío de un trabajo de gcloud CLI con el conector de Spanner.

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

   Reemplaza lo siguiente:

   1. CLUSTER_NAME Es el nombre del clúster nuevo.
   2. REGION: Es una región de Compute Engine disponible para ejecutar la carga de trabajo.
   3. CONNECTOR_VERSION: Versión del conector de Spanner. Elige la versión del conector de Spanner en la lista de versiones del repositorio de GitHub GoogleCloudDataproc/spark-spanner-connector.

Trabajo de spark-submit

1. Conéctate a la instancia principal del clúster de Managed Service para Apache Spark con SSH.
   1. Ve a la página Clústeres de Managed Service para Apache Spark en la consola de Google Cloud y, luego, haz clic en el nombre de tu clúster.
   2. En la página Detalles del clúster, selecciona la pestaña Instancias de VM. Luego, haz clic en SSH a la derecha del nombre del nodo instancia principal del clúster.

      

      Se abrirá una ventana del navegador en tu directorio principal del nodo.

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

2. Crea un archivo singers.py en el nodo principal con el editor de texto vi, vim o nano instalado previamente.
   1. Pega el siguiente código en el archivo singers.py después de completar las variables de marcador de posición en el archivo singers.py. Ten en cuenta que la función Data Boost de Spanner está habilitada, lo que tiene un impacto casi nulo en la instancia principal de 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()
        

      Reemplaza lo siguiente:

      1. PROJECT_ID: Es el ID del proyecto de Google Cloud . Los IDs del proyecto se enumeran en la sección Información del proyecto del panel de la consola de Google Cloud .
      2. INSTANCE_ID, DATABASE_ID y TABLE_NAME : Consulta Cómo configurar una instancia de Spanner con la tabla de base de datos Singers.
   2. Guarda el archivo singers.py.
3. Ejecuta singers.py con spark-submit para crear la tabla Singers de Spanner.

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

   Reemplaza lo siguiente:

   1. CONNECTOR_VERSION: Versión del conector de Spanner. Elige la versión del conector de Spanner en la lista de versiones del repositorio de GitHub GoogleCloudDataproc/spark-spanner-connector.

   El resultado es el siguiente:

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

Para ejecutar el código de ejemplo de Scala en tu clúster, completa los siguientes pasos:

1. Conéctate a la instancia principal del clúster de Managed Service para Apache Spark con SSH.
   1. Ve a la página Clústeres de Managed Service para Apache Spark en la consola de Google Cloud y, luego, haz clic en el nombre de tu clúster.
   2. En la página Detalles del clúster, selecciona la pestaña Instancias de VM. Luego, haz clic en SSH a la derecha del nombre del nodo instancia principal del clúster.

      Se abrirá una ventana del navegador en tu directorio principal del nodo.

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

2. Crea un archivo singers.scala en el nodo principal con el editor de texto vi, vim o nano instalado previamente.
   1. Pega el siguiente código en el archivo singers.scala. Ten en cuenta que la función Data Boost de Spanner está habilitada, lo que tiene un impacto casi nulo en la instancia principal de 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()
        }
      }
        

      Reemplaza lo siguiente:

      1. PROJECT_ID: Es el ID del proyecto de Google Cloud . Los IDs del proyecto se enumeran en la sección Información del proyecto del panel de la consola de Google Cloud .
      2. INSTANCE_ID, DATABASE_ID y TABLE_NAME : Consulta Cómo configurar una instancia de Spanner con la tabla de base de datos Singers.
   2. Guarda el archivo singers.scala.
3. Inicia el REPL de spark-shell.

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

   Reemplaza lo siguiente:

   CONNECTOR_VERSION: Versión del conector de Spanner. Elige la versión del conector de Spanner en la lista de versiones del repositorio de GitHub GoogleCloudDataproc/spark-spanner-connector.

4. Ejecuta singers.scala con el comando :load singers.scala para crear la tabla Singers de Spanner. La lista de salida muestra ejemplos del resultado 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)
     

Cómo leer gráficos de Spanner

El conector de Spanner admite la exportación del gráfico a DataFrames separados de nodos y bordes, así como la exportación directamente a GraphFrames.

En el siguiente ejemplo, se exporta un Spanner a un GraphFrame. Utiliza la clase SpannerGraphConnector de Python, incluida en el archivo JAR del conector de Spanner, para leer el gráfico de Spanner.

Nota: Completa las variables de marcador de posición antes de ejecutar el ejemplo.

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

Reemplaza lo siguiente:

• CONNECTOR_VERSION: Versión del conector de Spanner. Elige la versión del conector de Spanner en la lista de versiones del repositorio de GitHub GoogleCloudDataproc/spark-spanner-connector.
• PROJECT_ID: Es el ID del proyecto de Google Cloud . Los IDs del proyecto se enumeran en la sección Información del proyecto del panel de la consola Google Cloud .
• INSTANCE_ID, DATABASE_ID y TABLE_NAME: Inserta los IDs de la instancia, la base de datos y el gráfico.

Para exportar nodos y bordes DataFrames en lugar de GraphFrames, usa load_dfs en su lugar:

df_vertices, df_edges, df_id_map = connector.load_dfs()

Consulta Exporting Spanner Graphs en GitHub para obtener más información.

Escribe tablas de Spanner

El conector de Spanner admite la escritura de un DataFrame de Spark en una tabla de Spanner con la API de fuentes de datos de Spark.

Ejemplo de escritura de DataFrame en una tabla de Spanner

Propaga las variables antes de guardar y ejecutar el código.

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

Reemplaza lo siguiente.

• PROJECT_ID: Es el ID del proyecto de Google Cloud . Los IDs del proyecto se enumeran en la sección Información del proyecto del panel de la consola Google Cloud .
• INSTANCE_ID, DATABASE_ID y TABLE_NAME: Inserta los IDs de la instancia, la base de datos y la tabla.

Nota: Para obtener más información, consulta Cómo escribir en tablas de Spanner en GitHub.

Realiza una limpieza

Para evitar que se apliquen cargos continuos a tu cuenta de Google Cloud , puedes detener o borrar tu clúster de Managed Service para Apache Spark y borrar tu instancia de Spanner.

¿Qué sigue?

• Consulta los ejemplos de pyspark.sql.DataFrame.
• Para obtener información sobre la compatibilidad de idiomas de los DataFrames de Spark, consulta lo siguiente:
  • Scala
  • Java
  • R
• Consulta el repositorio del conector de Spark Spanner en GitHub.
• Consulta las sugerencias de ajuste para trabajos de Spark.