MongoDB

操作步骤

1. 准备以下测试代码和依赖包来访问MongoDB，并将此测试代码和依赖包分别编译打包生成jar包上传至您的OSS。

   测试代码示例：

   package com.aliyun.spark
   
   import com.mongodb.spark.MongoSpark
   import com.mongodb.spark.config.{ReadConfig, WriteConfig}
   import org.apache.spark.SparkConf
   import org.apache.spark.sql.SparkSession
   import org.bson.Document
   
   object SparkOnMongoDB {
   
     def main(args: Array[String]): Unit = {
       //获取MongoDB的connectionStringURI，database和collection。
       val connectionStringURI = args(0)
       val database = args(1)
       val collection = args(2)
       //Spark侧的表名。
       var sparkTableName = if (args.size > 3) args(3) else "spark_on_mongodb_sparksession_test01"
   
       val sparkSession = SparkSession
         .builder()
         .appName("scala spark on MongoDB test")
         .getOrCreate()
   
       //Spark读取MongoDB数据有多种方式。
       //使用Dataset API方式：
       //设置MongoDB的参数。
       val sparkConf = new SparkConf()
         .set("spark.mongodb.input.uri", connectionStringURI)
         .set("spark.mongodb.input.database", database)
         .set("spark.mongodb.input.collection", collection)
         .set("spark.mongodb.output.uri", connectionStringURI)
         .set("spark.mongodb.output.database", database)
         .set("spark.mongodb.output.collection", collection)
       val readConf = ReadConfig(sparkConf)
       //获取Dataframe。
       val df = MongoSpark.load(sparkSession, readConf)
       df.show(1)
   
       //使用MongoSpark.save入库数据到MongoDB。
       val docs =
         """
           |{"id": "id105", "name": "name105"}
           |{"id": "id106", "name": "name106"}
           |{"id": "id107", "name": "name107"}
           |"""
           .trim.stripMargin.split("[\\r\\n]+").toSeq
       val writeConfig: WriteConfig = WriteConfig(Map(
         "uri" -> connectionStringURI,
         "spark.mongodb.output.database" -> database,
         "spark.mongodb.output.collection"-> collection))
       MongoSpark.save(sparkSession.sparkContext.parallelize(docs.map(Document.parse)), writeConfig)
   
       //使用Sql的方式，SQL的方式有两种，指定Schema和不指定Schema。
       //指定Schema的创建方式，Schema中的字段必须和MongoDB中Collection的Schema一致。
       var createCmd =
       s"""CREATE TABLE ${sparkTableName} (
          |      id String,
          |      name String
          |    ) USING com.mongodb.spark.sql
          |    options (
          |    uri '$connectionStringURI',
          |    database '$database',
          |    collection '$collection'
          |    )""".stripMargin
   
       sparkSession.sql(createCmd)
       var querySql = "select * from " + sparkTableName + " limit 1"
       sparkSession.sql(querySql).show
   
       //不指定Schema的创建方式，不指定Schema，Spark会映射MOngoDB中collection的Schema。
       sparkTableName = sparkTableName + "_noschema"
       createCmd =
         s"""CREATE TABLE ${sparkTableName} USING com.mongodb.spark.sql
            |    options (
            |    uri '$connectionStringURI',
            |    database '$database',
            |    collection '$collection'
            |    )""".stripMargin
   
       sparkSession.sql(createCmd)
       querySql = "select * from " + sparkTableName + " limit 1"
       sparkSession.sql(querySql).show
   
       sparkSession.stop()
   
     }
   }

   MongoDB依赖的pom文件：

           <dependency>
               <groupId>org.mongodb.spark</groupId>
               <artifactId>mongo-spark-connector_2.11</artifactId>
               <version>2.4.2</version>
           </dependency>
           <dependency>
               <groupId>org.mongodb</groupId>
               <artifactId>mongo-java-driver</artifactId>
               <version>3.8.2</version>
           </dependency>

2. 登录Data Lake Analytics管理控制台。

3. 在页面左上角，选择MongoDB实例所在地域。

4. 单击左侧导航栏中的Serverless Spark > 作业管理。

5. 在作业编辑页面，单击创建作业。

6. 在创建作业模板页面，按照页面提示进行参数配置后，单击确定创建Spark作业。

   

7. 单击Spark作业名，在Spark作业编辑框中输入以下作业内容，并按照以下参数说明进行参数值替换。保存并提交Spark作业。

   {
       "args": [
           "mongodb://root:xxx@xxx:3717,xxx:3717/xxx",  #MongoDB集群中的“连接信息 (Connection StringURI)”。
           "config",  #MongoDB集群中的数据库名称。
           "test_collection",  #MongoDB集群中的collection名称。
           "spark_on_mongodb"  #Spark中创建映射MongoDB中collection的表名。
       ],
       "file": "oss://spark_test/jars/mongodb/spark-examples-0.0.1-SNAPSHOT.jar",  #存放测试软件包的OSS路径。
       "name": "mongodb-test",
       "jars": [
           "oss://spark_test/jars/mongodb/mongo-java-driver-3.8.2.jar",  ##存放测试软件依赖包的OSS路径。
           "oss://spark_test/jars/mongodb/mongo-spark-connector_2.11-2.4.2.jar"  ##存放测试软件依赖包的OSS路径。
       ],
       "className": "com.aliyun.spark.SparkOnMongoDB",
       "conf": {
           "spark.driver.resourceSpec": "small",
           "spark.executor.instances": 2,
           "spark.executor.resourceSpec": "small",
           "spark.dla.eni.enable": "true", 
           "spark.dla.eni.vswitch.id": "vsw-xxx",   #可访问MongoDB的交换机id。
           "spark.dla.eni.security.group.id": "sg-xxx"  #可访问MongoDB的安全组id。
       }
   }