# java样例代码
#### 开发说明
mongo只支持增强型跨源。只能使用包年包月队列。
![](https://support.huaweicloud.com/devg-dli/public_sys-resources/note_3.0-zh-cn.png)
DDS即文档数据库服务，兼容MongoDB协议。
- 前提条件 在DLI管理控制台上已完成创建增强跨源连接，并绑定包年/包月队列。具体操作请参考《[数据湖探索用户指南](https://support.huaweicloud.com/usermanual-dli/dli_01_0426.html)》。
  ![](https://support.huaweicloud.com/devg-dli/public_sys-resources/note_3.0-zh-cn.png)
  认证用的password硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。
  
- 代码实现详解
  1. 导入依赖
     - 涉及到的mvn依赖库
       ```
       <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-sql_2.11</artifactId>
         <version>2.3.2</version>
       </dependency>
       ```
       
     
     - import相关依赖包
       ```
       import org.apache.spark.SparkConf;
       import org.apache.spark.SparkContext;
       import org.apache.spark.api.java.JavaRDD;
       import org.apache.spark.api.java.JavaSparkContext;
       import org.apache.spark.sql.Dataset;
       import org.apache.spark.sql.Row;
       import org.apache.spark.sql.SQLContext;
       import org.apache.spark.sql.SaveMode;
       ```
       
      
  
  2. 创建会话
     ```
     SparkContext sparkContext = new SparkContext(new SparkConf().setAppName("datasource-mongo"));
     JavaSparkContext javaSparkContext = new JavaSparkContext(sparkContext);
     SQLContext sqlContext = new SQLContext(javaSparkContext);
     ```
     
   
- 通过DataFrame API 访问
  1. 读取json数据为DataFrame
     ```
     JavaRDD<String> javaRDD = javaSparkContext.parallelize(Arrays.asList("{\"id\":\"5\",\"name\":\"Ann\",\"age\":\"23\"}"));
     Dataset<Row> dataFrame = sqlContext.read().json(javaRDD);
     ```
     
  
  2. 设置连接参数
     ```
     String url = "192.168.4.62:8635,192.168.5.134:8635/test?authSource=admin";
     String uri = "mongodb://username:pwd@host:8635/db";
     String user = "rwuser";
     String database = "test";
     String collection = "test";
     String password = "######";
     ```
     ![](https://support.huaweicloud.com/devg-dli/public_sys-resources/note_3.0-zh-cn.png)
     详细的参数说明请参考[表1](https://support.huaweicloud.com/devg-dli/dli_09_0114.html#dli_09_0114__zh-cn_topic_0204096844_table2072415395012)。
     
  
  3. 导入数据到mongo
     ```
     dataFrame.write().format("mongo")   
          .option("url",url)
          .option("uri",uri)     
          .option("database",database)  
          .option("collection",collection)   
          .option("user",user)    
          .option("password",password)  
          .mode(SaveMode.Overwrite)  
          .save();
     ```
     
  
  4. 读取mongo上的数据
     ```
     sqlContext.read().format("mongo")  
         .option("url",url)    
         .option("uri",uri)
         .option("database",database)   
         .option("collection",collection)  
         .option("user",user)   
         .option("password",password)  
         .load().show();
     ```
     
  
  5. 操作结果 ![](https://support.huaweicloud.com/devg-dli/zh-cn_image_0265883908.png)
     
   
- 提交Spark作业
  1. 将写好的java代码文件上传至OBS桶中。
  
  2. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。
     ![](https://support.huaweicloud.com/devg-dli/public_sys-resources/note_3.0-zh-cn.png)
     - 如果选择spark版本为2.3.2（即将下线）或2.4.5提交作业时，需要指定Module模块，名称为：sys.datasource.mongo。
     
     - 如果选择Spark版本为3.1.1及以上版本时，无需选择Module模块， 需在 'Spark参数（--conf)' 配置 spark.driver.extraClassPath=/usr/share/extension/dli/spark-jar/datasource/mongo/\*
       spark.executor.extraClassPath=/usr/share/extension/dli/spark-jar/datasource/mongo/\*
       
     
     - 通过控制台提交作业请参考《[数据湖探索用户指南](https://support.huaweicloud.com/usermanual-dli/dli_01_0384.html)》中的"选择依赖资源参数说明"。
     
     - 通过API提交作业请参考《数据湖探索API参考》\>《[创建批处理作业](https://support.huaweicloud.com/api-dli/dli_02_0124.html)》中"表2-请求参数说明"关于"modules"参数的说明。
       
   
 
#### 完整示例代码
```
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.SaveMode;
import java.util.Arrays;
public class TestMongoSparkSql {
  public static void main(String[] args) {
    SparkContext sparkContext = new SparkContext(new SparkConf().setAppName("datasource-mongo"));
    JavaSparkContext javaSparkContext = new JavaSparkContext(sparkContext);
    SQLContext sqlContext = new SQLContext(javaSparkContext);
//    // Read json file as DataFrame, read csv / parquet file, same as json file distribution
//    DataFrame dataFrame = sqlContext.read().format("json").load("filepath");
    // Read RDD in JSON format to create DataFrame
    JavaRDD<String> javaRDD = javaSparkContext.parallelize(Arrays.asList("{\"id\":\"5\",\"name\":\"Ann\",\"age\":\"23\"}"));
    Dataset<Row> dataFrame = sqlContext.read().json(javaRDD);
    String url = "192.168.4.62:8635,192.168.5.134:8635/test?authSource=admin";
    String uri = "mongodb://username:pwd@host:8635/db";
    String user = "rwuser";
    String database = "test";
    String collection = "test";
    String password = "######";
    dataFrame.write().format("mongo")
            .option("url",url)
            .option("uri",uri)
            .option("database",database) 
            .option("collection",collection)
            .option("user",user)
            .option("password",password)
            .mode(SaveMode.Overwrite) 
            .save();
    sqlContext.read().format("mongo")
            .option("url",url)
            .option("uri",uri)
            .option("database",database)
            .option("collection",collection)
            .option("user",user)
            .option("password",password)
            .load().show();
    sparkContext.stop();
    javaSparkContext.close();
  }
}
```