# scala样例代码
#### 操作场景
本例提供使用Spark作业访问DWS数据源的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跨源访问DWS的关联表。
```
sparkSession.sql(
"CREATE TABLE IF NOT EXISTS dli_to_dws USING JDBC OPTIONS (
'url'='jdbc:postgresql://to-dws-1174404209-cA37siB6.datasource.com:8000/postgres',
'dbtable'='customer',
'user'='dbadmin',
'passwdauth'='######'//DLI侧创建的Password类型的跨源认证名称。使用跨源认证则无需在作业中配置账号和密码。
)"
)
```
表1创建表参数
| 参数 | 说明 |
|:---|:---|
| url | DWS的连接地址，需要先创建跨源连接，管理控制台操作请参考《数据湖探索用户指南》。 创建经典型跨源连接后，使用经典型跨源连接中返回的连接地址。 创建增强型跨源连接后，可以使用DWS提供的"JDBC连接字符串（内网）"，或者内网地址和内网端口访问，格式为"协议头://内网IP:内网端口/数据库名"，例如："jdbc:postgresql://192.168.0.77:8000/postgres"，获取方式请参考"图 DWS集群信息"。 说明： DWS的连接地址格式为："协议头://访问地址:访问端口/数据库名" 例如： jdbc:postgresql://to-dws-1174405119-ihlUr78j.datasource.com:8000/postgres 如果想要访问DWS中自定义数据库，请在这个连接里将"postgres"修改为对应的数据库名字。 |
| passwdauth | DLI侧创建的Password类型的跨源认证名称。使用跨源认证则无需在作业中配置账号和密码。 |
| dbtable | 数据库postgres中的数据表。 |
| 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"。 |
图1DWS集群信息
2. 插入数据
```
sparkSession.sql("insert into dli_to_dws values(1, 'John',24),(2, 'Bob',32)")
```
3. 查询数据
```
val dataFrame = sparkSession.sql("select * from dli_to_dws")
dataFrame.show()
```
插入数据前：
插入数据后：
4. 删除关联表
```
sparkSession.sql("drop table dli_to_dws")
```
#### 通过DataFrame API访问数据源
1. 连接配置。
```
val url = "jdbc:postgresql://to-dws-1174405057-EA1Kgo8H.datasource.com:8000/postgres"
val username = "dbadmin"
val password = "######"
val dbtable = "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. 导入数据到DWS。
```
df.write.format("jdbc")
.option("url", url)
.option("dbtable", dbtable)
.option("user", username)
.option("password", password)
.mode(SaveMode.Append)
.save()
```
SaveMode 有四种保存类型：
- ErrorIfExists：如果已经存在数据，则抛出异常。
- Overwrite：如果已经存在数据，则覆盖原数据。
- Append：如果已经存在数据，则追加保存。
- Ignore：如果已经存在数据，则不做操作。这类似于SQL中的"如果不存在则创建表"。
4. 读取DWS上的数据。
- 方式一：read.format()方法
```
val jdbcDF = sparkSession.read.format("jdbc")
.option("url", url)
.option("dbtable", dbtable)
.option("user", username)
.option("password", password)
.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()方法读取到的dateFrame注册为临时表，就可使用sql语句进行数据查询了。
```
jdbcDF.registerTempTable("customer_test")
sparkSession.sql("select * from customer_test where id = 1").show()
```
查询结果：
#### DataFrame相关操作
createDataFrame() 方法创建的数据和read.format() 方法及read.jdbc() 方法查询的数据都为DataFrame对象，可以直接进行查询单条记录等操作（在"[通过DataFrame API访问数据源]"中，提到将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()
```
#### 提交作业
1. 将写好的代码生成jar包，上传至OBS桶中。
2. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。
- 如果选择spark版本为2.3.2（即将下线）或2.4.5提交作业时，需要指定Module模块，名称为：sys.datasource.dws。
- 如果选择Spark版本为3.1.1及以上版本时，无需选择Module模块， 需在 'Spark参数（--conf)' 配置 spark.driver.extraClassPath=/usr/share/extension/dli/spark-jar/datasource/dws/\*
spark.executor.extraClassPath=/usr/share/extension/dli/spark-jar/datasource/dws/\*
- 通过控制台提交作业请参考《[数据湖探索用户指南](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访问
认证用的password硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。
```
import java.util.Properties
import org.apache.spark.sql.SparkSession
object Test_SQL_DWS {
def main(args: Array[String]): Unit = {
// Create a SparkSession session.
val sparkSession = SparkSession.builder().getOrCreate()
// Create a data table for DLI-associated DWS
sparkSession.sql("CREATE TABLE IF NOT EXISTS dli_to_dws USING JDBC OPTIONS (
'url'='jdbc:postgresql://to-dws-1174405057-EA1Kgo8H.datasource.com:8000/postgres',
'dbtable'='customer',
'user'='dbadmin',
'password'='######')")
//*****************************SQL model***********************************
//Insert data into the DLI data table
sparkSession.sql("insert into dli_to_dws values(1,'John',24),(2,'Bob',32)")
//Read data from DLI data table
val dataFrame = sparkSession.sql("select * from dli_to_dws")
dataFrame.show()
//drop table
sparkSession.sql("drop table dli_to_dws")
sparkSession.close()
}
}
```
- 通过DataFrame API访问
认证用的password硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。
```
import java.util.Properties
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SaveMode
object Test_SQL_DWS {
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:postgresql://to-dws-1174405057-EA1Kgo8H.datasource.com:8000/postgres"
val username = "dbadmin"
val password = "######"
val dbtable = "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 dws_table_1 table
df.write.format("jdbc")
.option("url", url)
.option("dbtable", dbtable)
.option("user", username)
.option("password", password)
.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 DWS using 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()
//Way two：Read data from DWS 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()
}
}
```