# scala样例代码
#### 开发说明
- 前提条件 在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 java.util.Properties
import org.apache.spark.sql.{Row,SparkSession}
import org.apache.spark.sql.SaveMode
```
2. 创建会话。
```
val sparkSession = SparkSession.builder().getOrCreate()
```
- 通过SQL API 访问
1. 创建DLI跨源访问 rds的关联表，填写连接参数。
```
sparkSession.sql(
"CREATE TABLE IF NOT EXISTS dli_to_rds USING JDBC OPTIONS (
'url'='jdbc:mysql://to-rds-1174404209-cA37siB6.datasource.com:3306', //根据实际url修改
'dbtable'='test.customer',
'user'='root', //根据实际user修改
'password'='######', //根据实际password修改
'driver'='com.mysql.jdbc.Driver')")
```
表1创建表参数
| 参数 | 说明 |
|:---|:---|
| url | RDS的连接地址，需要先创建跨源连接，管理控制台操作请参考《数据湖探索用户指南》。 创建经典型跨源连接后，使用经典型跨源连接中返回的连接地址。 创建增强型跨源连接后，使用RDS提供的"内网域名"或者内网地址和数据库端口访问，MySQL格式为"协议头://内网IP:内网端口"，PostGre格式为"协议头://内网IP:内网端口/数据库名"。 例如："jdbc:mysql://192.168.0.193:3306"或者"jdbc:postgresql://192.168.0.193:5432/postgres"，获取方式请参考"图 RDS集群信息"。 说明： 经典型跨源的连接地址默认格式为："协议头://访问地址:访问端口" 例如：jdbc:mysql://to-rds-1174405119-oLRHAGE7.datasource.com:3306 如果想要访问RDS的postgre集群，需要将连接地址中的协议头修改为"jdbc:postgresql"，并在连接地址最后加上"/数据库名"。 例如：jdbc:postgresql://to-rds-1174405119-oLRHAGE7.datasource.com:3306/postgreDB |
| dbtable | 访问MySQL集群填写"数据库名.表名"，访问PostGre集群填写"模式名.表名"。 说明： 如果数据库和表不存在，请先创建数据库和表，否则系统会报错并且运行失败。 |
| user | RDS数据库用户名。 |
| password | RDS数据库用户名对应密码。 |
| driver | jdbc驱动类名，访问MySQL集群请填写："com.mysql.jdbc.Driver"，访问PostGre集群请填写："org.postgresql.Driver"。 |
| partitionColumn | 读取数据时，用于设置并发使用的数值型字段。 说明： - "partitionColumn"，"lowerBound"，"upperBound"，"numPartitions"4个参数必须同时设置，不支持仅设置其中一部分 - 为了提升并发读取的性能，建议使用自增列。 |
| lowerBound | partitionColumn设置的字段数据最小值，该值包含在返回结果中。 |
| upperBound | partitionColumn设置的字段数据最大值，该值不包含在返回结果中。 |
| numPartitions | 读取数据时并发数。 说明： 实际读取数据时，会根据lowerBound与upperBound，平均分配给每个task获取其中一部分的数据。例如： 'partitionColumn'='id', 'lowerBound'='0', 'upperBound'='100', 'numPartitions'='2' DLI中会起2个并发task，一个task执行id\>=0 and id \< 50，另一个task执行id \>=50 and id \< 100。 |
| fetchsize | 读取数据时，每一批次获取数据的记录数，默认值1000。设置越大性能越好，但占用内存越多，该值设置过大会有内存溢出的风险。 |
| batchsize | 写入数据时，每一批次写入数据的记录数，默认值1000。设置越大性能越好，但占用内存越多，该值设置过大会有内存溢出的风险。 |
| truncate | 执行overwrite时是否不删除原表，直接执行清空表操作，取值范围： - true - false 默认为'false'，即在执行overwrite操作时，先将原表删除再重新建表。 |
| isolationLevel | 事务隔离级别，取值范围： - NONE - READ_UNCOMMITTED - READ_COMMITTED - REPEATABLE_READ - SERIALIZABLE 默认值为'READ_UNCOMMITTED'。 |
图1RDS集群信息
2. 插入数据
```
sparkSession.sql("insert into dli_to_rds values(1, 'John',24),(2, 'Bob',32)")
```
3. 查询数据
```
val dataFrame = sparkSession.sql("select * from dli_to_rds")
dataFrame.show()
```
插入数据前：
插入数据后：
4. 删除关联表
```
sparkSession.sql("drop table dli_to_rds")
```
- 通过DataFrame API 访问
1. 连接参数配置
```
val url = "jdbc:mysql://to-rds-1174405057-EA1Kgo8H.datasource.com:3306"
val username = "root"
val password = "######"
val dbtable = "test.customer"
```
2. 创建DataFrame，添加数据，并重命名字段。
```
var dataFrame_1 = sparkSession.createDataFrame(List((8, "Jack_1", 18)))
val df = dataFrame_1.withColumnRenamed("_1", "id")
.withColumnRenamed("_2", "name")
.withColumnRenamed("_3", "age")
```
3. 导入数据到RDS。
```
df.write.format("jdbc")
.option("url", url)
.option("dbtable", dbtable)
.option("user", username)
.option("password", password)
.option("driver", "com.mysql.jdbc.Driver")
.mode(SaveMode.Append)
.save()
```
SaveMode 有四种保存类型：
- ErrorIfExists：如果已经存在数据，则抛出异常。
- Overwrite：如果已经存在数据，则覆盖原数据。
- Append：如果已经存在数据，则追加保存。
- Ignore：如果已经存在数据，则不做操作。这类似于SQL中的"如果不存在则创建表"。
4. 读取RDS上的数据。
- 方式一：read.format()方法
```
val jdbcDF = sparkSession.read.format("jdbc")
.option("url", url)
.option("dbtable", dbtable)
.option("user", username)
.option("password", password)
.option("driver", "org.postgresql.Driver")
.load()
```
- 方式二：read.jdbc()方法
```
val properties = new Properties()
properties.put("user", username)
properties.put("password", password)
val jdbcDF2 = sparkSession.read.jdbc(url, dbtable, properties)
```
插入数据前：
插入数据后：
使用上述read.format()或者read.jdbc()方法读取到的dataFrame注册为临时表，就可使用sql语句进行数据查询了。
```
jdbcDF.registerTempTable("customer_test")
sparkSession.sql("select * from customer_test where id = 1").show()
```
查询结果：
- DataFrame相关操作 createDataFrame() 方法创建的数据和read.format() 方法及read.jdbc() 方法查询的数据都为DataFrame对象，可以直接进行查询单条记录等操作（在"步骤[4]"中，提到将DataFrame数据注册为临时表）。
- where where 方法中可传入包含and 和 or 的条件筛选表达式，返回过滤后的DataFrame对象，示例如下：
```
jdbcDF.where("id = 1 or age <=10").show()
```
- filter filter同where的使用方式一致，传入条件筛选表达式，返回过滤后的结果 。示例如下：
```
jdbcDF.filter("id = 1 or age <=10").show()
```
- select 传入待查询的字段，返回指定字段的DataFrame对象，并且可多个字段查询，示例如下：
- 示例1：
```
jdbcDF.select("id").show()
```
- 示例2：
```
jdbcDF.select("id", "name").show()
```
- 示例3：
```
jdbcDF.select("id","name").where("id<4").show()
```
- selectExpr 对字段进行特殊处理。例如，可使用selectExpr修改字段名。示例如下：
将name字段取名name_test，age数据加1。
```
jdbcDF.selectExpr("id", "name as name_test", "age+1").show()
```
- col 获取指定字段。不同于select，col每次只能获取一个字段，返回类型为Column类型，示例如下：
```
val idCol = jdbcDF.col("id")
```
- drop 删除指定字段。传入要删除的字段，返回不包含此字段的DataFrame对象，并且每次只能删除一个字段，示例如下：
```
jdbcDF.drop("id").show()
```
- 提交Spark作业
1. 将写好的代码生成jar包，上传至OBS桶中。
2. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。
- 如果选择spark版本为2.3.2（即将下线）或2.4.5提交作业时，需要指定Module模块，名称为：sys.datasource.rds。
- 如果选择Spark版本为3.1.1及以上版本时，无需选择Module模块， 需在 'Spark参数（--conf)' 配置 spark.driver.extraClassPath=/usr/share/extension/dli/spark-jar/datasource/rds/\*
spark.executor.extraClassPath=/usr/share/extension/dli/spark-jar/datasource/rds/\*
- 通过控制台提交作业请参考《[数据湖探索用户指南](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 java.util.Properties
import org.apache.spark.sql.SparkSession
object Test_SQL_RDS {
def main(args: Array[String]): Unit = {
// Create a SparkSession session.
val sparkSession = SparkSession.builder().getOrCreate()
// Create a data table for DLI-associated RDS
sparkSession.sql("CREATE TABLE IF NOT EXISTS dli_to_rds USING JDBC OPTIONS (
'url'='jdbc:mysql://to-rds-1174404209-cA37siB6.datasource.com:3306,
'dbtable'='test.customer',
'user'='root',
'password'='######',
'driver'='com.mysql.jdbc.Driver')")
//*****************************SQL model***********************************
//Insert data into the DLI data table
sparkSession.sql("insert into dli_to_rds values(1,'John',24),(2,'Bob',32)")
//Read data from DLI data table
val dataFrame = sparkSession.sql("select * from dli_to_rds")
dataFrame.show()
//drop table
sparkSession.sql("drop table dli_to_rds")
sparkSession.close()
}
}
```
- 通过DataFrame API访问
```
import java.util.Properties
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SaveMode
object Test_SQL_RDS {
def main(args: Array[String]): Unit = {
// Create a SparkSession session.
val sparkSession = SparkSession.builder().getOrCreate()
//*****************************DataFrame model***********************************
// Set the connection configuration parameters. Contains url, username, password, dbtable.
val url = "jdbc:mysql://to-rds-1174404209-cA37siB6.datasource.com:3306"
val username = "root"
val password = "######"
val dbtable = "test.customer"
// Create a DataFrame and initialize the DataFrame data.
var dataFrame_1 = sparkSession.createDataFrame(List((1, "Jack", 18)))
// Rename the fields set by the createDataFrame() method.
val df = dataFrame_1.withColumnRenamed("_1", "id")
.withColumnRenamed("_2", "name")
.withColumnRenamed("_3", "age")
// Write data to the rds_table_1 table
df.write.format("jdbc")
.option("url", url)
.option("dbtable", dbtable)
.option("user", username)
.option("password", password)
.option("driver", "com.mysql.jdbc.Driver")
.mode(SaveMode.Append)
.save()
// DataFrame object for data manipulation
//Filter users with id=1
var newDF = df.filter("id!=1")
newDF.show()
// Filter the id column data
var newDF_1 = df.drop("id")
newDF_1.show()
// Read the data of the customer table in the RDS database
// Way one：Read data from RDS using read.format()
val jdbcDF = sparkSession.read.format("jdbc")
.option("url", url)
.option("dbtable", dbtable)
.option("user", username)
.option("password", password)
.option("driver", "com.mysql.jdbc.Driver")
.load()
// Way two：Read data from RDS using read.jdbc()
val properties = new Properties()
properties.put("user", username)
properties.put("password", password)
val jdbcDF2 = sparkSession.read.jdbc(url, dbtable, properties)
/**
* Register the dateFrame read by read.format() or read.jdbc() as a temporary table, and query the data
* using the sql statement.
*/
jdbcDF.registerTempTable("customer_test")
val result = sparkSession.sql("select * from customer_test where id = 1")
result.show()
sparkSession.close()
}
}
```
- DataFrame相关操作
```
// The where() method uses " and" and "or" for condition filters, returning filtered DataFrame objects
jdbcDF.where("id = 1 or age <=10").show()
// The filter() method is used in the same way as the where() method.
jdbcDF.filter("id = 1 or age <=10").show()
// The select() method passes multiple arguments and returns the DataFrame object of the specified field.
jdbcDF.select("id").show()
jdbcDF.select("id", "name").show()
jdbcDF.select("id","name").where("id<4").show()
/**
* The selectExpr() method implements special handling of fields, such as renaming, increasing or
* decreasing data values.
*/
jdbcDF.selectExpr("id", "name as name_test", "age+1").show()
// The col() method gets a specified field each time, and the return type is a Column type.
val idCol = jdbcDF.col("id")
/**
* The drop() method returns a DataFrame object that does not contain deleted fields, and only one field
* can be deleted at a time.
*/
jdbcDF.drop("id").show()
```