scala样例代码

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

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

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()

1. 创建DLI跨源访问 rds的关联表，填写连接参数。

   1 2 3 4 5 6 7

   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:3306/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'。

   图1 RDS集群信息
   
2. 插入数据

   1	sparkSession.sql("insert into dli_to_rds values(1, 'John',24),(2, 'Bob',32)")

3. 查询数据

   1 2	val dataFrame = sparkSession.sql("select * from dli_to_rds") dataFrame.show()

   插入数据前：

   

   插入数据后：

   

4. 删除关联表

   1	sparkSession.sql("drop table dli_to_rds")

1. 连接参数配置

   1 2 3 4

   val url = "jdbc:mysql://to-rds-1174405057-EA1Kgo8H.datasource.com:3306" val username = "root" val password = "######" val dbtable = "test.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. 导入数据到RDS。

   1 2 3 4 5 6 7 8

   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 有四种保存类型：

   • ErrorIfExis：如果已经存在数据，则抛出异常。
   • Overwrite：如果已经存在数据，则覆盖原数据。
   • Append：如果已经存在数据，则追加保存。
   • Ignore：如果已经存在数据，则不做操作。这类似于SQL中的“如果不存在则创建表”。
4. 读取RDS上的数据。
   • 方式一：read.format()方法

     1 2 3 4 5 6 7

     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()方法

     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()方法读取到的dataFrame注册为临时表，就可使用sql语句进行数据查询了。

   1 2	jdbcDF.registerTempTable("customer_test") sparkSession.sql("select * from customer_test where id = 1").show()

   查询结果：

   

createDataFrame() 方法创建的数据和read.format() 方法及read.jdbc() 方法查询的数据都为DataFrame对象，可以直接进行查询单条记录等操作（在“步骤4”中，提到将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包，上传至DLI中。

   控制台操作请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>《上传资源包》。

2. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。

   控制台操作请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>《创建批处理作业》。

   

   • 如果选择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/*

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