# scala样例代码
#### 开发说明
mongo只支持增强型跨源。只能使用包年包月队列。
DDS即文档数据库服务，兼容MongoDB协议。
在DLI管理控制台上已完成创建增强跨源连接，并绑定包年/包月队列。具体操作请参考《[数据湖探索用户指南](https://support.huaweicloud.com/usermanual-dli/dli_01_0426.html)》。
认证用的password硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。
- 构造依赖信息，创建SparkSession
1. 导入依赖。
涉及到mvn依赖
```
org.apache.spark
spark-sql_2.11
2.3.2
```
import相关依赖包
```
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
```
创建session
```
val sparkSession = SparkSession.builder().appName("datasource-mongo").getOrCreate()
```
- 通过SQL API访问
1. 创建DLI跨源访问 mongo的关联表
```
sparkSession.sql(
"create table test_dds(id string, name string, age int) using mongo options(
'url' = '192.168.4.62:8635,192.168.5.134:8635/test?authSource=admin',
'uri' = 'mongodb://username:pwd@host:8635/db',
'database' = 'test',
'collection' = 'test',
'user' = 'rwuser',
'password' = '######')")
```
表1创建表参数
| 参数 | 说明 |
|:---|:---|
| url | - url的格式为： "IP:PORT\[,IP:PORT\]/\[DATABASE\]\[.COLLECTION\]\[AUTH_PROPERTIES\]" 例如： ``` "192.168.4.62:8635/test?authSource=admin" ``` - url需要在mongo（DDS）的连接地址的截取得到。 获取到的mongo的连接地址格式为："协议头://用户名:密码@访问地址:访问端口/数据库名?authSource=admin" 例如： ``` mongodb://rwuser:****@192.168.4.62:8635,192.168.5.134:8635/test?authSource=admin ``` |
| uri | uri的格式为：mongodb://username:pwd@host:8635/db 其中以下参数需要修改为实际值： - "username"为创建的mongo（DDS）数据库用户名。 - "pwd"为创建的mongo（DDS）数据库用户名对应的密码。 - "host"为创建的mongo（DDS）数据库实例IP。 - "db"为创建的mongo（DDS）数据库名称。 mongo（DDS）数据库用户创建详见：[创建DDS数据库账户](https://support.huaweicloud.com/usermanual-dds/dds_03_0057.html)。 |
| database | DDS的数据库名，如果在"url"中同时指定了数据库名，则"url"中的数据库名不生效。 |
| collection | DDS中的collection名，如果在"url"中同时指定了collection，则"url"中的collection不生效。 说明： 如果在DDS中已存在collection，则建表可以不指定schema信息，DLI会根据collection中的数据自动生成schema信息。 |
| user | 访问DDS集群用户名。 |
| password | 访问DDS集群密码。 |
2. 插入数据
```
sparkSession.sql("insert into test_dds values('3', 'Ann',23)")
```
3. 查询数据
```
sparkSession.sql("select * from test_dds").show()
```
操作结果
- 通过DataFrame API访问
1. 设置连接参数
```
val url = "192.168.4.62:8635,192.168.5.134:8635/test?authSource=admin"
val uri = "mongodb://username:pwd@host:8635/db"
val user = "rwuser"
val database = "test"
val collection = "test"
val password = "######"
```
2. 构造schema
```
val schema = StructType(List(StructField("id", StringType), StructField("name", StringType), StructField("age", IntegerType)))
```
3. 构造DataFrame
```
val rdd = spark.sparkContext.parallelize(Seq(Row("1", "John", 23), Row("2", "Bob", 32)))
val dataFrame = spark.createDataFrame(rdd, schema)
```
4. 导入数据到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()
```
保存类型：Overwrite、Append、ErrorIfExists、Ignore 四种。
5. 读取mongo上的数据
```
val jdbcDF = spark.read.format("mongo").schema(schema)
.option("url", url)
.option("uri", uri)
.option("database", database)
.option("collection", collection)
.option("user", user)
.option("password", password)
.load()
```
操作结果
- 提交Spark作业
1. 将写好的代码生成jar包，上传至OBS桶中。
2. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。 控制台操作请参考《[数据湖探索用户指南](https://support.huaweicloud.com/usermanual-dli/dli_01_0384.html)》。API操作请参考《数据湖探索API参考》\>《[创建批处理作业](https://support.huaweicloud.com/api-dli/dli_02_0124.html)》。
- 如果选择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"参数的说明。
#### 完整示例代码
- Maven依赖
```
org.apache.spark
spark-sql_2.11
2.3.2
```
- 通过SQL API访问
```
import org.apache.spark.sql.SparkSession
object TestMongoSql {
def main(args: Array[String]): Unit = {
val sparkSession = SparkSession.builder().getOrCreate()
sparkSession.sql(
"create table test_dds(id string, name string, age int) using mongo options(
'url' = '192.168.4.62:8635,192.168.5.134:8635/test?authSource=admin',
'uri' = 'mongodb://username:pwd@host:8635/db',
'database' = 'test',
'collection' = 'test',
'user' = 'rwuser',
'password' = '######')")
sparkSession.sql("insert into test_dds values('3', 'Ann',23)")
sparkSession.sql("select * from test_dds").show()
sparkSession.close()
}
}
```
- 通过DataFrame API访问
```
import org.apache.spark.sql.{Row, SaveMode, SparkSession}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
object Test_Mongo_SparkSql {
def main(args: Array[String]): Unit = {
// Create a SparkSession session.
val spark = SparkSession.builder().appName("mongodbTest").getOrCreate()
// Set the connection configuration parameters.
val url = "192.168.4.62:8635,192.168.5.134:8635/test?authSource=admin"
val uri = "mongodb://username:pwd@host:8635/db"
val user = "rwuser"
val database = "test"
val collection = "test"
val password = "######"
// Setting up the schema
val schema = StructType(List(StructField("id", StringType), StructField("name", StringType), StructField("age", IntegerType)))
// Setting up the DataFrame
val rdd = spark.sparkContext.parallelize(Seq(Row("1", "John", 23), Row("2", "Bob", 32)))
val dataFrame = spark.createDataFrame(rdd, schema)
// Write data to 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()
// Reading data from mongo
val jdbcDF = spark.read.format("mongo").schema(schema)
.option("url", url)
.option("uri", uri)
.option("database", database)
.option("collection", collection)
.option("user", user)
.option("password", password)
.load()
jdbcDF.show()
spark.close()
}
}
```