2.10 使用 Spark 作业跨源访问数据源
2.10.6 对接 RDS
}
2.10.5.4 故障处理
运行 Spark 作业，作业运行失败，作业日志中提示 No respond 错误
● 问题现象
运行Spark作业，作业运行失败，作业日志中提示No respond错误
● 解决方案
重新创建Spark作业，创建作业时需要在“Spark参数（--conf）”中添加配置：
“spark.sql.mrs.opentsdb.ssl.enabled=true”。
2.10.6 对接 RDS
2.10.6.1 scala 样例代码 开发说明
● 前提条件
在DLI管理控制台上已完成创建跨源连接并绑定队列。具体操作请参考《数据湖探 索用户指南》。
● 构造依赖信息，创建SparkSession a. 导入依赖
涉及到的mvn依赖库
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.3.2</version>
</dependency>
import相关依赖包
import java.util.Properties
import org.apache.spark.sql.{Row,SparkSession}
import org.apache.spark.sql.SaveMode
b. 创建会话。
val sparkSession = SparkSession.builder().getOrCreate()
● 通过SQL API 访问
a. 创建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')")
表2-28 创建表参数
参数 说明
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"。
partitionCol
umn 读取数据时，用于设置并发使用的数值型字段。
说明
● “partitionColumn”，“lowerBound”，“upperBound”，
“numPartitions”4个参数必须同时设置，不支持仅设置其中 一部分
● 为了提升并发读取的性能，建议使用自增列。
lowerBound partitionColumn设置的字段数据最小值，该值包含在返回 结果中。
upperBound partitionColumn设置的字段数据最大值，该值不包含在返 回结果中。
参数 说明 numPartitio
ns 读取数据时并发数。
说明实际读取数据时，会根据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操作时，先将原表删除 再重新建表。
isolationLeve
l 事务隔离级别，取值范围：
● NONE
● READ_UNCOMMITTED
● READ_COMMITTED
● REPEATABLE_READ
● SERIALIZABLE
默认值为'READ_UNCOMMITTED'。
图2-44 RDS 集群信息
b. 插入数据
sparkSession.sql("insert into dli_to_rds values(1, 'John',24),(2, 'Bob',32)")
c. 查询数据
val dataFrame = sparkSession.sql("select * from dli_to_rds") dataFrame.show()
插入数据前：
插入数据后：
d. 删除关联表
sparkSession.sql("drop table dli_to_rds")
● 通过DataFrame API 访问 a. 连接参数配置
val url = "jdbc:mysql://to-rds-1174405057-EA1Kgo8H.datasource.com:3306"
val username = "root"
val password = "######"
val dbtable = "test.customer"
b. 创建DataFrame，添加数据，并重命名字段。
var dataFrame_1 = sparkSession.createDataFrame(List((8, "Jack_1", 18))) val df = dataFrame_1.withColumnRenamed("_1", "id")
.withColumnRenamed("_2", "name") .withColumnRenamed("_3", "age")
c. 导入数据到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 有四种保存类型：
● ErrorIfExis：如果已经存在数据，则抛出异常。
● Overwrite：如果已经存在数据，则覆盖原数据。
● Append：如果已经存在数据，则追加保存。
● Ignore：如果已经存在数据，则不做操作。这类似于SQL中的“如果不存在则创 建表”。
d. 读取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对象，可以直接进行查询单条记录等操作（在“步骤d”
中，提到将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作业
a. 将写好的代码生成jar包，上传至DLI中。控制台操作请参考《数据湖探索用户 指南》。API操作请参考《数据湖探索API参考》>《上传资源包》。
b. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。控制台操作 请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>
《创建批处理作业》。
说明
● 提交访问MySQL作业时，需要指定Module模块，名称为：sys.datasource.rds。
● 通过控制台提交作业请参考《数据湖探索用户指南》中的“选择依赖资源参数说 明”表说明。
● 通过API提交作业请参考《数据湖探索API参考》>《创建批处理作业》中“表2-请 求参数说明”关于“modules”参数的说明。
完整示例代码
● Maven依赖
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.3.2</version>
</dependency>
● 通过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 // Way one：Read data from RDS using read.format() val jdbcDF = sparkSession.read.format("jdbc") .option("url", url)
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()
2.10.6.2 pyspark 样例代码 开发说明
from __future__ import print_function
from pyspark.sql.types import StructType, StructField, IntegerType, StringType from pyspark.sql import SparkSession
b. 创建会话
sparkSession = SparkSession.builder.appName("datasource-rds").getOrCreate()
● 通过DataFrame API 访问 a. 连接参数配置
url = "jdbc:mysql://to-rds-1174404952-ZgPo1nNC.datasource.com:3306"
dbtable = "test.customer"
user = "root"
password = "######"
driver = "com.mysql.jdbc.Driver"
参数说明请参考表2-28。
b. 设置数据
dataList = sparkSession.sparkContext.parallelize([(123, "Katie", 19)])
c. 设置schema
schema = StructType([StructField("id", IntegerType(), False), StructField("name", StringType(), False), StructField("age", IntegerType(), False)])
d. 创建DataFrame
dataFrame = sparkSession.createDataFrame(dataList, schema)
e. 保存数据到RDS
jdbcDF = sparkSession.read \ .format("jdbc") \ .option("url", url) \ .option("dbtable", dbtable) \ .option("user", user) \
.option("password", password) \ .option("driver", driver) \
"CREATE TABLE IF NOT EXISTS dli_to_rds USING JDBC OPTIONS ( 'url'='jdbc:mysql://to-rds-1174404952-ZgPo1nNC.datasource.com:3306', 'dbtable'='test.customer',
'user'='root',
'password'='######',
'driver'='com.mysql.jdbc.Driver')")
创建表参数请参考表2-28。
b. 插入数据
sparkSession.sql("insert into dli_to_rds values(3,'John',24)")
c. 查询数据
jdbcDF_after = sparkSession.sql("select * from dli_to_rds") jdbcDF_after.show()
d. 操作结果
● 提交Spark作业
a. 将写好的python代码文件上传至DLI中。控制台操作请参考《数据湖探索用户 指南》。API操作请参考《数据湖探索API参考》>《上传资源包》。
b. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。控制台操作 请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>
《创建批处理作业》。
c. 创建Spark作业完成后，在控制台单击右上角“执行”提交作业，页面显示
“批处理作业提交成功”说明Spark作业提交成功，可以在Spark作业管理页 面查看提交的作业的状态和日志。
说明
● 创建Spark作业时选择的“所属队列”为创建跨源连接时所绑定的队列。
● 创建Spark作业时，选择的Module模块，名称为：sys.datasource.rds。
● 通过控制台提交作业请参考《数据湖探索用户指南》中的“选择依赖资源参数说 明”表说明。
● 通过API提交作业请参考《数据湖探索API参考》>《创建批处理作业》中“表2-请 求参数说明”关于“modules”参数的说明。
完整示例代码
说明
直接复制如下样例代码到py文件中后，需要注意文件内容中的“\”后面可能会有unexpected character的问题。需要将“\”后面的缩进或是空格全部删除。
● 通过DataFrame API访问
# _*_ coding: utf-8 _*_
from __future__ import print_function
from pyspark.sql.types import StructType, StructField, IntegerType, StringType from pyspark.sql import SparkSession
if __name__ == "__main__":
# Create a SparkSession session.
sparkSession = SparkSession.builder.appName("datasource-rds").getOrCreate() # Set cross-source connection parameters.
url = "jdbc:mysql://to-rds-1174404952-ZgPo1nNC.datasource.com:3306"
dbtable = "test.customer"
user = "root"
password = "######"
driver = "com.mysql.jdbc.Driver"
# Create a DataFrame and initialize the DataFrame data.
dataList = sparkSession.sparkContext.parallelize([(123, "Katie", 19)]) # Setting schema
schema = StructType([StructField("id", IntegerType(), False), StructField("name", StringType(), False), StructField("age", IntegerType(), False)])
# Create a DataFrame from RDD and schema
dataFrame = sparkSession.createDataFrame(dataList, schema)
# Write data to the RDS.
jdbcDF.show() # close session sparkSession.stop()
● 通过SQL API访问
# _*_ coding: utf-8 _*_
from __future__ import print_function from pyspark.sql import SparkSession if __name__ == "__main__":
# Create a SparkSession session.
sparkSession = SparkSession.builder.appName("datasource-rds").getOrCreate() # Createa data table for DLI - associated RDS
sparkSession.sql(
"CREATE TABLE IF NOT EXISTS dli_to_rds USING JDBC OPTIONS ( 'url'='jdbc:mysql://to-rds-1174404952-ZgPo1nNC.datasource.com:3306', 'dbtable'='test.customer',
'user'='root',
'password'='######',
'driver'='com.mysql.jdbc.Driver')") # Insert data into the DLI data table
sparkSession.sql("insert into dli_to_rds values(3,'John',24)") # Read data from DLI data table
jdbcDF = sparkSession.sql("select * from dli_to_rds") jdbcDF.show()
# close session sparkSession.stop()
2.10.6.3 java 样例代码
开发说明
▪
涉及到的mvn依赖库<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.3.2</version>
</dependency>
▪
import相关依赖包import org.apache.spark.sql.SparkSession;
b. 创建会话
SparkSession sparkSession = SparkSession.builder().appName("datasource-rds").getOrCreate();
● 通过SQL API 访问
– 创建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修改
'url'='jdbc:mysql://to-rds-1174404209-cA37siB6.datasource.com:3306', //根据实际url修改