AnalyticDB:Access a MongoDB data source

Step 1: Download the required JAR packages

Download the following JAR packages from Maven Central and upload them to an OSS bucket:

Step 2: Add the Maven dependency

Add the following dependency to your pom.xml:

<dependency>
  <groupId>org.mongodb.spark</groupId>
  <artifactId>mongo-spark-connector_2.12</artifactId>
  <version>10.1.1</version>
</dependency>

Step 3: Write and package the Spark program

Write a Spark program that connects to ApsaraDB for MongoDB, compile it, and package it as a JAR file. The example below reads from a MongoDB collection and prints the results. This topic uses spark-mongodb.jar as the output JAR name.

package com.aliyun.spark

import org.apache.spark.sql.SparkSession

object SparkOnMongoDB {
  def main(args: Array[String]): Unit = {
    // The VPC endpoint of the ApsaraDB for MongoDB instance.
    // Find it on the Database Connection page in the ApsaraDB for MongoDB console.
    val connectionUri = args(0)
    // The name of the database.
    val database = args(1)
    // The name of the collection.
    val collection = args(2)

    val spark = SparkSession.builder()
      .appName("MongoSparkConnectorIntro")
      .config("spark.mongodb.read.connection.uri", connectionUri)
      .config("spark.mongodb.write.connection.uri", connectionUri)
      .getOrCreate()

    val df = spark.read.format("mongodb")
      .option("database", database)
      .option("collection", collection)
      .load()

    df.show()

    spark.stop()
  }
}

For additional configuration options, see Configuration Options in the MongoDB Spark Connector documentation. Code examples are available for writing to MongoDB and reading from MongoDB.

Step 4: Submit the Spark JAR job

1. Log on to the AnalyticDB for MySQL console. In the upper-left corner, select a region. In the left navigation pane, click Clusters, then click the target cluster ID.

2. In the left navigation pane, choose Job Development > Spark JAR Development.

3. In the editor, enter the following job configuration:

   Important

   AnalyticDB for MySQL Spark supports both VPC and internet access to ApsaraDB for MongoDB. Use VPC for better security and performance.

   Parameter	Description
   args	Arguments passed to the JAR program: the MongoDB VPC endpoint URI, database name, and collection name, in that order. Separate multiple arguments with commas.
   file	OSS path of your compiled application JAR (spark-mongodb.jar).
   jars	OSS paths of the connector JAR packages that Spark needs to access MongoDB.
   name	Name of the Spark job.
   className	Entry class of the Java or Scala program. Not required for Python applications.
   spark.adb.eni.enabled	Specifies whether to enable Elastic Network Interface (ENI) access. You must enable ENI access when you use Data Lakehouse Edition Spark to access the MongoDB data source.
   spark.adb.eni.vswitchId	The vSwitch ID. Find it on the Basic Information page of your ApsaraDB for MongoDB instance.
   spark.adb.eni.securityGroupId	ID of the security group added to the ApsaraDB for MongoDB instance. If no security group exists, see Add a security group.
   Other conf parameters	Spark job configuration parameters, following the same format as Apache Spark. Use key:value format, separated by commas. See Spark application configuration parameters.

   {
     "args": [
       "mongodb://<username>:<password>@<host1>:<port1>,<host2>:<port2>,...,<hostN>:<portN>/<database_name>",
       "<database_name>",
       "<collection_name>"
     ],
     "file": "oss://<bucket_name>/spark-mongodb.jar",
     "jars": [
       "oss://<bucket_name>/mongo-spark-connector_2.12-10.1.1.jar",
       "oss://<bucket_name>/mongodb-driver-sync-4.8.2.jar",
       "oss://<bucket_name>/bson-4.8.2.jar",
       "oss://<bucket_name>/bson-record-codec-4.8.2.jar",
       "oss://<bucket_name>/mongodb-driver-core-4.8.2.jar"
     ],
     "name": "MongoSparkConnectorIntro",
     "className": "com.aliyun.spark.SparkOnMongoDB",
     "conf": {
       "spark.driver.resourceSpec": "medium",
       "spark.executor.instances": 2,
       "spark.executor.resourceSpec": "medium",
       "spark.adb.eni.enabled": "true",
       "spark.adb.eni.vswitchId": "vsw-bp14pj8h0****",
       "spark.adb.eni.securityGroupId": "sg-bp11m93k021tp****"
     }
   }

   Replace all <placeholder> values with your actual resource identifiers. The following table describes each parameter:

4. Click Execute Now.

5. {
     "args": [
   	  "mongodb://root:ALiyun***@dds-bp13f15dfdeac6141.mongodb.rds.aliyuncs.com:3717,dds-bp13f15dfdeac6142.mongodb.rds.aliyuncs.com:3717/admin?replicaSet=mgset-68693516",
       "test",
       "mongo"
   	],
     "file": "oss://adb-test1/MongDB/spark-mongodb-1.0-SNAPSHOT.jar",
   	"jars": [
   		"oss://adb-test1/MongDB/mongo-spark-connector_2.12-10.1.1.jar",
   	  "oss://adb-test1/MongDB/mongodb-driver-sync-4.8.2.jar",
   	  "oss://adb-test1/MongDB/bson-4.8.2.jar",
   	  "oss://adb-test1/MongDB/bson-record-codec-4.8.2.jar",
   	  "oss://adb-test1/MongDB/mongodb-driver-core-4.8.2.jar"
   	],
     "name": "MongoSparkConnectorIntro",
   	"className": "com.aliyun.spark.SparkOnMongoDB",
     "conf": {
       "spark.driver.resourceSpec": "medium",
       "spark.executor.instances": 2,
       "spark.executor.resourceSpec": "medium",
       "spark.adb.eni.enabled": "true",
   	  "spark.adb.eni.vswitchId": "vsw-bp1skxmyj1lhdx4sq2u0p",
   	  "spark.adb.eni.securityGroupId": "sg-bp15it2qvpa017gkun5j"
     }
   }