计算
弹性云服务器 ECS
Flexus云服务
裸金属服务器 BMS
弹性伸缩 AS
镜像服务 IMS
专属主机 DeH
函数工作流 FunctionGraph
云手机服务器 CPH
Huawei Cloud EulerOS
网络
虚拟私有云 VPC
弹性公网IP EIP
虚拟专用网络 VPN
弹性负载均衡 ELB
NAT网关 NAT
云专线 DC
VPC终端节点 VPCEP
云连接 CC
企业路由器 ER
企业交换机 ESW
全球加速 GA
安全与合规
安全技术与应用
Web应用防火墙 WAF
企业主机安全 HSS
云防火墙 CFW
安全云脑 SecMaster
DDoS防护 AAD
数据加密服务 DEW
数据库安全服务 DBSS
云堡垒机 CBH
数据安全中心 DSC
云证书管理服务 CCM
边缘安全 EdgeSec
态势感知 SA
威胁检测服务 MTD
CDN与智能边缘
内容分发网络 CDN
CloudPond云服务
智能边缘云 IEC
迁移
主机迁移服务 SMS
对象存储迁移服务 OMS
云数据迁移 CDM
迁移中心 MGC
大数据
MapReduce服务 MRS
数据湖探索 DLI
表格存储服务 CloudTable
云搜索服务 CSS
数据接入服务 DIS
数据仓库服务 GaussDB(DWS)
数据治理中心 DataArts Studio
数据可视化 DLV
数据湖工厂 DLF
湖仓构建 LakeFormation
企业应用
云桌面 Workspace
应用与数据集成平台 ROMA Connect
云解析服务 DNS
专属云
专属计算集群 DCC
IoT物联网
IoT物联网
设备接入 IoTDA
智能边缘平台 IEF
用户服务
账号中心
费用中心
成本中心
资源中心
企业管理
工单管理
国际站常见问题
ICP备案
我的凭证
支持计划
客户运营能力
合作伙伴支持计划
专业服务
区块链
区块链服务 BCS
Web3节点引擎服务 NES
解决方案
SAP
高性能计算 HPC
视频
视频直播 Live
视频点播 VOD
媒体处理 MPC
实时音视频 SparkRTC
数字内容生产线 MetaStudio
存储
对象存储服务 OBS
云硬盘 EVS
云备份 CBR
存储容灾服务 SDRS
高性能弹性文件服务 SFS Turbo
弹性文件服务 SFS
云硬盘备份 VBS
云服务器备份 CSBS
数据快递服务 DES
专属分布式存储服务 DSS
容器
云容器引擎 CCE
容器镜像服务 SWR
应用服务网格 ASM
华为云UCS
云容器实例 CCI
管理与监管
云监控服务 CES
统一身份认证服务 IAM
资源编排服务 RFS
云审计服务 CTS
标签管理服务 TMS
云日志服务 LTS
配置审计 Config
资源访问管理 RAM
消息通知服务 SMN
应用运维管理 AOM
应用性能管理 APM
组织 Organizations
优化顾问 OA
IAM 身份中心
云运维中心 COC
资源治理中心 RGC
应用身份管理服务 OneAccess
数据库
云数据库 RDS
文档数据库服务 DDS
数据管理服务 DAS
数据复制服务 DRS
云数据库 GeminiDB
云数据库 GaussDB
分布式数据库中间件 DDM
数据库和应用迁移 UGO
云数据库 TaurusDB
人工智能
人脸识别服务 FRS
图引擎服务 GES
图像识别 Image
内容审核 Moderation
文字识别 OCR
AI开发平台ModelArts
图像搜索 ImageSearch
对话机器人服务 CBS
华为HiLens
视频智能分析服务 VIAS
语音交互服务 SIS
应用中间件
分布式缓存服务 DCS
API网关 APIG
微服务引擎 CSE
分布式消息服务Kafka版
分布式消息服务RabbitMQ版
分布式消息服务RocketMQ版
多活高可用服务 MAS
事件网格 EG
企业协同
华为云会议 Meeting
云通信
消息&短信 MSGSMS
云生态
合作伙伴中心
云商店
开发者工具
SDK开发指南
API签名指南
Terraform
华为云命令行工具服务 KooCLI
其他
产品价格详情
系统权限
管理控制台
客户关联华为云合作伙伴须知
消息中心
公共问题
开发与运维
应用管理与运维平台 ServiceStage
软件开发生产线 CodeArts
需求管理 CodeArts Req
部署 CodeArts Deploy
性能测试 CodeArts PerfTest
编译构建 CodeArts Build
流水线 CodeArts Pipeline
制品仓库 CodeArts Artifact
测试计划 CodeArts TestPlan
代码检查 CodeArts Check
代码托管 CodeArts Repo
云应用引擎 CAE
开天aPaaS
云消息服务 KooMessage
云手机服务 KooPhone
云空间服务 KooDrive
本文导读

scala样例代码

更新时间:2025-01-10 GMT+08:00

开发说明

  • 前提条件

    在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跨源访问 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集群信息”。

      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")
      
  • 通过DataFrame API 访问
    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()
      

      查询结果:

  • DataFrame相关操作

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

  • 提交Spark作业
    1. 将写好的代码生成jar包,上传至OBS桶中。
    2. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。
      说明:
      • 如果选择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”参数的说明。

完整示例代码

  • 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访问
     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
    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访问
     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
    69
    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
        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 RDS using read.format()
        val jdbcDF = sparkSession.read.format("jdbc")
                        .option("url", url)
                        .option("dbtable", dbtable)
                        .option("user", username)
                        .option("password", password)
                        .option("driver", "com.mysql.jdbc.Driver")
                        .load()
        // Way two:Read data from RDS 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()
      }
    }
    
  • DataFrame相关操作
     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
      // 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()
    

我们使用cookie来确保您的高速浏览体验。继续浏览本站,即表示您同意我们使用cookie。 详情

文档反馈

文档反馈

意见反馈

0/500

标记内容

同时提交标记内容