訪問MongoDB資料來源

前提條件

• AnalyticDB for MySQL叢集的產品系列為企業版、基礎版或湖倉版。

• 已建立資料庫帳號。

  • 如果是通過阿里雲帳號訪問，只需建立高許可權帳號。

  • 如果是通過RAM使用者訪問，需要建立高許可權帳號和普通帳號並且將RAM使用者綁定到普通帳號上。

• 已建立Job型資源群組。

• 已建立與AnalyticDB for MySQL叢集同地區的ApsaraDB for MongoDB，並在MongoDB中建立資料庫、建立集合和寫入資料。具體操作，請參見MongoDB入門概述。

• 已將ApsaraDB for MongoDB的交換器IP添加到MongoDB的白名單中。

  說明

  在ApsaraDB for MongoDB控制台的基本信息頁面查看交換器ID。登入專用網路管理主控台，查看目標交換器的IP。

• 已將ApsaraDB for MongoDB添加到安全性群組中，且安全性群組規則的入方向與出方向允許存取MongoDB連接埠的訪問請求。具體操作，請參見添加安全性群組和添加安全性群組規則。

操作步驟

1. 下載AnalyticDB for MySQLSpark訪問ApsaraDB for MongoDB依賴的Jar包。下載連結mongo-spark-connector_2.12-10.1.1.jar、mongodb-driver-sync-4.8.2.jar、bson-4.8.2.jar、bson-record-codec-4.8.2.jar和mongodb-driver-core-4.8.2.jar。

2. 在pom.xml檔案的dependencies中添加依賴項。

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

3. 編寫如下樣本程式訪問ApsaraDB for MongoDB，並進行編譯打包。本文產生的Jar包名稱為spark-mongodb.jar。範例程式碼如下：

   package com.aliyun.spark
   
   import org.apache.spark.sql.SparkSession
   
   object SparkOnMongoDB {
     def main(args: Array[String]): Unit = {
       // 訪問MongoDB的專用網路地址，可在MongoDB控制台的資料庫連接頁面查看。
       val connectionUri = args(0)
       // MongoDB資料庫名稱。
       val database = args(1)
       // MongoDB集合名稱。
       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()
     }
   }

   說明

   Spark訪問MongoDB的更多配置請參見Configuration Options，更多代碼請參見Write to MongoDB和Read from MongoDB。

4. 登入雲原生資料倉儲AnalyticDB MySQL控制台，在左上方選擇叢集所在地區。在左側導覽列，單擊集群清單，然後單擊目的地組群ID。

5. 在左側導覽列，單擊作業開發 > Spark Jar 開發。

6. 在編輯器視窗上方，選擇Job型資源群組和作業類型。本文以Batch類型為例。

7. 在編輯器中輸入以下作業內容。

   重要

   • AnalyticDB for MySQLSpark訪問ApsaraDB for MongoDB，支援通過專用網路和公網訪問。

   • 建議使用專用網路訪問。

   {
     "args": [
       -- 訪問MongoDB的專用網路地址，可在MongoDB控制台的資料庫連接頁面查看。
   	  "mongodb://<username>:<password>@<host1>:<port1>,<host2>:<port2>,...,<hostN>:<portN>/<database_name>",
       -- MongoDB資料庫的名稱。
       "<database_name>",
       -- MongoDB集合名稱。
       "<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****"
     }
   }

   參數說明如下。

   參數	說明
   args	請根據業務需求，填寫使用Jar包時需要的參數。多個參數之間以英文逗號（,）分隔。
   file	樣本程式spark-mongodb.jar所在的OSS路徑。
   jars	Spark訪問MongoDB依賴的Jar包所在的OSS路徑。
   name	Spark作業名稱。
   className	Java或者Scala程式入口類名稱。Python不需要指定入口類。
   spark.adb.eni.enabled	是否開啟ENI訪問。 使用湖倉版Spark訪問MongoDB資料來源時，需要開啟ENI訪問。
   spark.adb.eni.vswitchId	交換器ID。在目標ApsaraDB for MongoDB控制台的基本信息頁面擷取交換器ID。
   spark.adb.eni.securityGroupId	ApsaraDB for MongoDB中添加的安全性群組ID。如未添加安全性群組，請參見添加安全性群組。
   conf其他參數	與開源Spark中的配置項基本一致，參數格式為key:value形式，多個參數之間以英文逗號（,）分隔。更多conf參數，請參見Conf配置參數。

8. 單擊運行。

9. 待應用列表中目標應用的狀態為已完成，您可以單擊操作列的日誌查看MongoDB表的資料。