2.10 使用 Spark 作业跨源访问数据源
2.10.3 对接 DWS
}
2.10.3 对接 DWS
2.10.3.1 scala 样例代码 开发说明
<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跨源访问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', 'password'='######')")
表2-23 创建表参数
参数 说明
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"修改 为对应的数据库名字。
user DWS数据仓库用户名。
passwor
d DWS数据仓库用户名对应密码。
dbtable 数据库postgres中的数据表。
partition
Column 读取数据时，用于设置并发使用的数值型字段。
说明
● “partitionColumn”，“lowerBound”，“upperBound”，
“numPartitions”4个参数必须同时设置，不支持仅设置其中一部 分。
● 为了提升并发读取的性能，建议使用自增列。
lowerBo
und partitionColumn设置的字段数据最小值，该值包含在返回结果 中。
upperBo
und partitionColumn设置的字段数据最大值，该值不包含在返回结 果中。
参数 说明 numParti
tions 读取数据时并发数。
说明实际读取数据时，会根据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。设置 越大性能越好，但占用内存越多，该值设置过大会有内存溢出 的风险。
batchsiz
e 写入数据时，每一批次写入数据的记录数，默认值1000。设置 越大性能越好，但占用内存越多，该值设置过大会有内存溢出 的风险。
truncate 执行overwrite时是否不删除原表，直接执行清空表操作，取值 范围：
● true
● false
默认为“false”，即在执行overwrite操作时，先将原表删除再 重新建表。
isolation
Level 事务隔离级别，取值范围：
● NONE
● READ_UNCOMMITTED
● READ_COMMITTED
● REPEATABLE_READ
● SERIALIZABLE
默认值为“READ_UNCOMMITTED”。
图2-38 DWS 集群信息
b. 插入数据
sparkSession.sql("insert into dli_to_dws values(1, 'John',24),(2, 'Bob',32)")
c. 查询数据
val dataFrame = sparkSession.sql("select * from dli_to_dws") dataFrame.show()
插入数据前：
插入数据后：
d. 删除关联表
sparkSession.sql("drop table dli_to_dws")
● 通过DataFrame API 访问 a. 连接配置。
val url = "jdbc:postgresql://to-dws-1174405057-EA1Kgo8H.datasource.com:8000/postgres"
val username = "dbadmin"
val password = "######"
val dbtable = "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. 导入数据到DWS。
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中的“如果不存在则创 建表”。
d. 读取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对象，可以直接进行查询单条记录等操作（在“步骤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参考》>
《创建批处理作业》。
完整示例代码
● 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_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访问
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)
// DataFrame object for data manipulation //Filter users with id=1 //Way one：Read data from DWS using read.format() val jdbcDF = sparkSession.read.format("jdbc") .option("url", url)
//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() }}
2.10.3.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-dws").getOrCreate()
● 通过 DataFrame API 访问 a. 连接参数配置
url = "jdbc:postgresql://to-dws-1174404951-W8W4cW8I.datasource.com:8000/postgres"
dbtable = "customer"
user = "dbadmin"
password = "######"
driver = "org.postgresql.Driver"
b. 设置数据
dataList = sparkSession.sparkContext.parallelize([(1, "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. 保存数据到DWS
.option("password", password) \ .option("driver", driver) \
jdbcDF = sparkSession.read \ .format("jdbc") \ .option("url", url) \
.option("dbtable", dbtable) \ .option("user", user) \
.option("password", password) \ .option("driver", driver) \
.load() jdbcDF.show()
g. 操作结果
● 通过SQL API 访问
a. 创建DLI跨源访问 dws 的关联表。
sparkSession.sql(
"CREATE TABLE IF NOT EXISTS dli_to_dws USING JDBC OPTIONS (
'url'='jdbc:postgresql://to-dws-1174404951-W8W4cW8I.datasource.com:8000/postgres', 'dbtable'='customer',
'user'='dbadmin', 'password'='######',
'driver'='org.postgresql.Driver')") 说明
建表参数详情可参考表2-23。
b. 插入数据
sparkSession.sql("insert into dli_to_dws values(2,'John',24)")
c. 查询数据
jdbcDF = sparkSession.sql("select * from dli_to_dws").show()
d. 操作结果
● 提交Spark作业
a. 将写好的python代码文件上传至DLI中。控制台操作请参考《数据湖探索用户 指南》。API操作请参考《数据湖探索API参考》>《上传资源包》。
b. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。控制台操作 请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>
《创建批处理作业》。
说明
● 提交作业时，需要指定Module模块，名称为：sys.datasource.dws。
● 通过控制台提交作业请参考《数据湖探索用户指南》中的“选择依赖资源参数说 明”表说明。
● 通过API提交作业请参考《数据湖探索API参考》>《创建批处理作业》中“表2-请 求参数说明”关于“modules”参数的说明。
完整示例代码
● 通过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-dws").getOrCreate() # Set cross-source connection parameters
url = "jdbc:postgresql://to-dws-1174404951-W8W4cW8I.datasource.com:8000/postgres"
dbtable = "customer"
user = "dbadmin"
password = "######"
driver = "org.postgresql.Driver"
# Create a DataFrame and initialize the DataFrame data.
dataList = sparkSession.sparkContext.parallelize([(1, "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 DWS table
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-dws").getOrCreate() # Createa data table for DLI - associated DWS
sparkSession.sql(
"CREATE TABLE IF NOT EXISTS dli_to_dws USING JDBC OPTIONS (
'url'='jdbc:postgresql://to-dws-1174404951-W8W4cW8I.datasource.com:8000/postgres', 'dbtable'='customer',
'user'='dbadmin', 'password'='######',
'driver'='org.postgresql.Driver')") # Insert data into the DLI data table
sparkSession.sql("insert into dli_to_dws values(2,'John',24)") # Read data from DLI data table
jdbcDF = sparkSession.sql("select * from dli_to_dws").show() # close session
sparkSession.stop()
2.10.3.3 java 样例代码 开发说明
● 前提条件
在DLI管理控制台上已完成创建跨源连接并绑定队列。具体操作请参考《数据湖探 索用户指南》。
● 代码实现 a. 导入依赖
▪
涉及到的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-dws").getOrCreate();
● 通过SQL API 访问
a. 创建DLI跨源访问DWS的关联表，填写连接参数。
sparkSession.sql("CREATE TABLE IF NOT EXISTS dli_to_dws USING JDBC OPTIONS ('url'='jdbc:postgresql://10.0.0.233:8000/
postgres','dbtable'='test','user'='dbadmin','password'='**')");
b. 插入数据
sparkSession.sql("insert into dli_to_dws values(3,'Liu'),(4,'Xie')");
c. 查询数据
sparkSession.sql("select * from dli_to_dws").show();
插入数据后：
● 提交Spark作业
a. 将写好的代码文件生成jar包，上传至DLI中。控制台操作请参考《数据湖探索 用户指南》。API操作请参考《数据湖探索API参考》>《上传资源包》。
b. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。控制台操作 请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>
《创建批处理作业》。
说明
● 提交作业时，需要指定Module模块，名称为：sys.datasource.dws。
● 通过控制台提交作业请参考《数据湖探索用户指南》中的“选择依赖资源参数说 明”表说明。
● 通过API提交作业请参考《数据湖探索API参考》>《创建批处理作业》中“表2-请 求参数说明”关于“modules”参数的说明。
完整示例代码
通过SQL API 访问DWS表
import org.apache.spark.sql.SparkSession;
public class java_dws {
public static void main(String[] args) {
SparkSession sparkSession = SparkSession.builder().appName("datasource-dws").getOrCreate();
sparkSession.sql("CREATE TABLE IF NOT EXISTS dli_to_dws USING JDBC OPTIONS
('url'='jdbc:postgresql://10.0.0.233:8000/postgres','dbtable'='test','user'='dbadmin','password'='**')");
//*****************************SQL model***********************************
//Insert data into the DLI data table
sparkSession.sql("insert into dli_to_dws values(3,'Liu'),(4,'Xie')");
//Read data from DLI data table
sparkSession.sql("select * from dli_to_dws").show();
//drop table
sparkSession.sql("drop table dli_to_dws");
sparkSession.close();
} }