文档首页 > > 开发指南> 使用Spark作业开发DLI跨源连接> 对接DWS-scala> 详细开发说明

详细开发说明

分享
更新时间: 2020/05/12 GMT+08:00

前提条件

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

构造依赖信息,创建SparkSession

创建语句API可参考创建语句

  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
    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'='######')")
    
    表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"修改为对应的数据库名字。

    user

    DWS数据仓库用户名。

    password

    DWS数据仓库用户名对应密码。

    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相关操作

请参考DataFrame相关操作

提交Spark作业

  1. 将写好的代码生成jar包,上传至DLI中。控制台操作请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>《上传资源包》。
  2. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。控制台操作请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>《创建会话》和《创建批处理作业》。
分享:

    相关文档

    相关产品

文档是否有解决您的问题?

提交成功!

非常感谢您的反馈,我们会继续努力做到更好!

反馈提交失败,请稍后再试!

*必选

请至少选择或填写一项反馈信息

字符长度不能超过200

提交反馈 取消

如您有其它疑问,您也可以通过华为云社区问答频道来与我们联系探讨

智能客服提问云社区提问