文档首页/ 数据湖探索 DLI/ 开发指南/ Spark Jar作业开发指南/ 使用Spark作业跨源访问数据源/ 对接DWS/ scala样例代码
更新时间：2025-01-10 GMT+08:00
查看PDF
分享

scala样例代码

操作场景

本例提供使用Spark作业访问DWS数据源的scala样例代码。

在DLI管理控制台上已完成创建跨源连接并绑定队列。具体操作请参考《数据湖探索用户指南》。

认证用的password硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。

操作前准备

构造依赖信息，创建SparkSession
  1. 导入依赖

    涉及到的mvn依赖库

    1
2
3
4
5
    		 
    <dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql_2.11</artifactId>
  <version>2.3.2</version>
</dependency>
    import相关依赖包 
    1
2
3
    		 
    import java.util.Properties
import org.apache.spark.sql.{Row,SparkSession}
import org.apache.spark.sql.SaveMode
  2. 创建会话。 
    1
    		 
    val sparkSession = SparkSession.builder().getOrCreate()

通过SQL API 访问数据源

  1. 创建DLI跨源访问DWS的关联表。 
    1
2
3
4
5
6
7
8
    		 
    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”。
    图1 DWS集群信息
  2. 插入数据 
    1
    		 
    sparkSession.sql("insert into dli_to_dws values(1, 'John',24),(2, 'Bob',32)")
  3. 查询数据 
    1
2
    		 
    val dataFrame = sparkSession.sql("select * from dli_to_dws")
dataFrame.show()

    插入数据前：

    插入数据后：

  4. 删除关联表 
    1
    		 
    sparkSession.sql("drop table dli_to_dws")

通过DataFrame API访问数据源

  1. 连接配置。 
    1
2
3
4
    		 
    val url = "jdbc:postgresql://to-dws-1174405057-EA1Kgo8H.datasource.com:8000/postgres"
val username = "dbadmin"
val password = "######"
val dbtable = "customer"
  2. 创建DataFrame，添加数据，并重命名字段。 
    1
2
3
4
    		 
    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。 
    1
2
3
4
5
6
7
    		 
    df.write.format("jdbc")
  .option("url", url)
  .option("dbtable", dbtable)
  .option("user", username)
  .option("password", password)
  .mode(SaveMode.Append)
  .save()

    SaveMode 有四种保存类型：

    • ErrorIfExis：如果已经存在数据，则抛出异常。
    • Overwrite：如果已经存在数据，则覆盖原数据。
    • Append：如果已经存在数据，则追加保存。
    • Ignore：如果已经存在数据，则不做操作。这类似于SQL中的“如果不存在则创建表”。
  4. 读取DWS上的数据。
    • 方式一：read.format()方法 
      1
2
3
4
5
6
      		 
      val jdbcDF = sparkSession.read.format("jdbc")
                 .option("url", url)
                 .option("dbtable", dbtable)
                 .option("user", username)
                 .option("password", password)
                 .load()
    • 方式二：read.jdbc()方法 
      1
2
3
4
      		 
      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语句进行数据查询了。

    1
2
    		 
    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对象，示例如下：

    1
    		 
    jdbcDF.where("id = 1 or age <=10").show()

  • filter

    filter同where的使用方式一致，传入条件筛选表达式，返回过滤后的结果 。示例如下：

    1
    		 
    jdbcDF.filter("id = 1 or age <=10").show()

  • select

    传入待查询的字段，返回指定字段的DataFrame对象，并且可多个字段查询，示例如下：

    • 示例1： 
      1
      		 
      jdbcDF.select("id").show()

    • 示例2： 
      1
      		 
      jdbcDF.select("id", "name").show()

    • 示例3： 
      1
      		 
      jdbcDF.select("id","name").where("id<4").show()

  • selectExpr

    对字段进行特殊处理。例如，可使用selectExpr修改字段名。示例如下：

    将name字段取名name_test，age数据加1。

    1
    		 
    jdbcDF.selectExpr("id", "name as name_test", "age+1").show()
  • col

    获取指定字段。不同于select，col每次只能获取一个字段，返回类型为Column类型，示例如下：

    1
    		 
    val idCol = jdbcDF.col("id")
  • drop

    删除指定字段。传入要删除的字段，返回不包含此字段的DataFrame对象，并且每次只能删除一个字段，示例如下：

    1
    		 
    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/*

    • 通过控制台提交作业请参考《数据湖探索用户指南》中的“选择依赖资源参数说明”。
    • 通过API提交作业请参考《数据湖探索API参考》>《创建批处理作业》中“表2-请求参数说明”关于“modules”参数的说明。

完整示例代码

  • Maven依赖 
    1
2
3
4
5
    		 
    <dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql_2.11</artifactId>
  <version>2.3.2</version>
</dependency>
  • 通过SQL API访问

    认证用的password硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
    		 
    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硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
    		 
    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()
  }
}
父主题： 对接DWS