更新时间:2024-07-04 GMT+08:00

scala样例代码

开发说明

  • 前提条件

    在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包,上传至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”参数的说明。

完整示例代码

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