Comenzar con el Spark Connector

Gestión de dependencias

Proporcione las dependencias de Spark Core, Spark SQL y MongoDB Spark Connector a su herramienta de gestión de dependencias.

A partir de la versión 3.2.0, Apache Spark es compatible tanto con Scala 2.12 como con 2.13. Spark 3.1.3 y las versiones anteriores solo son compatibles con Scala 2.12. Para proporcionar soporte para ambas versiones de Scala, la versión 10.7.0 del Spark Connector produce dos artefactos:

• org.mongodb.spark:mongo-spark-connector_2.12:10.7.0 se compila contra Scala 2.12 y es compatible con Spark 3.1.x y versiones superiores.

• org.mongodb.spark:mongo-spark-connector_2.13:10.7.0 se compila contra Scala 2.13 y es compatible con Spark 3.2.x y versiones superiores.

El siguiente extracto de un archivo Maven pom.xml muestra cómo incluir dependencias compatibles con Scala 2.12:

<dependencies>
  <dependency>
    <groupId>org.mongodb.spark</groupId>
    <artifactId>mongo-spark-connector_2.12</artifactId>
    <version>10.7.0</version>
  </dependency>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.12</artifactId>
    <version>3.3.1</version>
  </dependency>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.12</artifactId>
    <version>3.3.1</version>
  </dependency>
</dependencies>

Configuración

Al especificar la configuración del Connector mediante SparkSession, se debe prefijar los ajustes adecuadamente. Para obtener detalles y otras opciones disponibles de MongoDB Spark Connector, consultar la guía Configuración de Spark.

package com.mongodb.spark_examples;
import org.apache.spark.sql.SparkSession;
public final class GettingStarted {
  public static void main(final String[] args) throws InterruptedException {
    /* Create the SparkSession.
     * If config arguments are passed from the command line using --conf,
     * parse args for the values to set.
     */
    SparkSession spark = SparkSession.builder()
      .master("local")
      .appName("MongoSparkConnectorIntro")
      .config("spark.mongodb.read.connection.uri", "mongodb://127.0.0.1/test.myCollection")
      .config("spark.mongodb.write.connection.uri", "mongodb://127.0.0.1/test.myCollection")
      .getOrCreate();
    // Application logic
  }
}

• El spark.mongodb.read.connection.uri especifica la dirección del servidor MongoDB (127.0.0.1), la base de datos a la que conectarse (test), la colección (myCollection) de la que leer datos y la preferencia de lectura.

• El spark.mongodb.write.connection.uri especifica la dirección del servidor MongoDB (127.0.0.1), la base de datos a la que se debe conectar (test) y la colección (myCollection) a la que se deben guardar los datos.

Puedes utilizar un SparkSession objeto para guardar datos en MongoDB, leer datos de MongoDB, crear conjuntos de datos y realizar operaciones SQL.

Shell de Python Spark

Este tutorial utiliza el shell pyspark, pero el código también funciona con aplicaciones autónomas de Python.

Al iniciar el shell pyspark, se puede especificar:

• la opción --packages para descargar el paquete de MongoDB Spark Connector. El siguiente paquete está disponible:

  • mongo-spark-connector

• la opción --conf para configurar el MongoDB Spark Connector. Estos ajustes configuran el objeto SparkConf.

  Nota

  Si utilizas SparkConf para configurar el Spark Connector, debes anteponer los ajustes adecuadamente. Para obtener detalles y otras opciones disponibles de MongoDB Spark Connector, consulta la guía Configurar Spark.

El siguiente ejemplo inicia el shell pyspark desde la línea de comandos:

./bin/pyspark --conf "spark.mongodb.read.connection.uri=mongodb://127.0.0.1/test.myCollection?readPreference=primaryPreferred" \
              --conf "spark.mongodb.write.connection.uri=mongodb://127.0.0.1/test.myCollection" \
              --packages org.mongodb.spark:mongo-spark-connector_2.12:10.7.0

• El spark.mongodb.read.connection.uri especifica la dirección del servidor MongoDB (127.0.0.1), la base de datos a la que se va a conectar (test), la colección (myCollection) de la que se van a leer los datos y la preferencia de lectura.

• El spark.mongodb.write.connection.uri especifica la dirección del servidor MongoDB (127.0.0.1), la base de datos a la que se conectará (test) y la colección (myCollection) a la que guardará los datos. Se conecta al puerto 27017 por defecto.

• La opción packages especifica las coordenadas Maven de Spark Connector, en el formato groupId:artifactId:version.

Los ejemplos de este tutorial utilizarán esta base de datos y colección.

Crea un objeto SparkSession

Si se especificaron las opciones de configuración spark.mongodb.read.connection.uri y spark.mongodb.write.connection.uri al iniciar pyspark, el objeto SparkSession por defecto las utiliza. Si prefiere crear su propio objeto SparkSession desde pyspark, puede utilizar SparkSession.builder y especificar diferentes opciones de configuración.

from pyspark.sql import SparkSession
my_spark = SparkSession \
    .builder \
    .appName("myApp") \
    .config("spark.mongodb.read.connection.uri", "mongodb://127.0.0.1/test.coll") \
    .config("spark.mongodb.write.connection.uri", "mongodb://127.0.0.1/test.coll") \
    .getOrCreate()

Se puede usar un objeto SparkSession para escribir datos en MongoDB, leer datos de MongoDB, crear DataFrames y realizar operaciones SQL.

Spark Shell

Al iniciar la shell de Spark, especifique:

• la opción --packages para descargar el paquete de MongoDB Spark Connector. El siguiente paquete está disponible:

  • mongo-spark-connector

• la opción --conf para configurar el MongoDB Spark Connector. Estos ajustes configuran el objeto SparkConf.

  Nota

  Si utilizas SparkConf para configurar el Spark Connector, debes anteponer los ajustes adecuadamente. Para obtener detalles y otras opciones disponibles de MongoDB Spark Connector, consulta la guía Configurar Spark.

Por ejemplo,

./bin/spark-shell --conf "spark.mongodb.read.connection.uri=mongodb://127.0.0.1/test.myCollection?readPreference=primaryPreferred" \
                  --conf "spark.mongodb.write.connection.uri=mongodb://127.0.0.1/test.myCollection" \
                  --packages org.mongodb.spark:mongo-spark-connector_2.12:10.7.0

• El spark.mongodb.read.connection.uri especifica la dirección del servidor MongoDB (127.0.0.1), la base de datos a la que se va a conectar (test), la colección (myCollection) de la que se van a leer los datos y la preferencia de lectura.

• El spark.mongodb.write.connection.uri especifica la dirección del servidor MongoDB (127.0.0.1), la base de datos a la que se conectará (test) y la colección (myCollection) a la que guardará los datos. Se conecta al puerto 27017 por defecto.

• La opción packages especifica las coordenadas Maven de Spark Connector, en el formato groupId:artifactId:version.

import com.mongodb.spark._

Aplicación autónoma de Scala

scalaVersion := "2.12",
libraryDependencies ++= Seq(
  "org.mongodb.spark" %% "mongo-spark-connector_2.12" % "10.7.0",
  "org.apache.spark" %% "spark-core" % "3.3.1",
  "org.apache.spark" %% "spark-sql" % "3.3.1"
)

package com.mongodb
object GettingStarted {
  def main(args: Array[String]): Unit = {
    /* Create the SparkSession.
     * If config arguments are passed from the command line using --conf,
     * parse args for the values to set.
     */
    import org.apache.spark.sql.SparkSession
    val spark = SparkSession.builder()
      .master("local")
      .appName("MongoSparkConnectorIntro")
      .config("spark.mongodb.read.connection.uri", "mongodb://127.0.0.1/test.myCollection")
      .config("spark.mongodb.write.connection.uri", "mongodb://127.0.0.1/test.myCollection")
      .getOrCreate()
  }
}

Solución de problemas

Si obtiene un java.net.BindException: Can't assign requested address,

• Verificar que no tenga otro shell de Spark ya en ejecución.

• Intentar configurar la variable de entorno SPARK_LOCAL_IP; por ejemplo,

  export SPARK_LOCAL_IP=127.0.0.1

• Intentar incluir la siguiente opción al iniciar la shell de Spark:

  --driver-java-options "-Djava.net.preferIPv4Stack=true"

Si aparecen errores al ejecutar los ejemplos de este tutorial, puede que se necesite borrar la caché local de Ivy (~/.ivy2/cache/org.mongodb.spark y ~/.ivy2/jars).