java样例代码

开发说明

• 前提条件

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

  

  认证用的password硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。

• 代码实现
  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	import org.apache.spark.sql.SparkSession;

  2. 创建会话

     1	SparkSession sparkSession = SparkSession.builder().appName("datasource-rds").getOrCreate();

• 通过SQL API 访问
  • 创建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。

  • 插入数据

    1	sparkSession.sql("insert into dli_to_rds values (1,'John',24)");

  • 查询数据

    1	sparkSession.sql("select * from dli_to_rd").show();

    插入数据后：

    

• 提交Spark作业
  1. 将写好的代码生成jar包，上传至DLI中。

     控制台操作请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>《上传资源包》。

  2. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。

     控制台操作请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>《创建批处理作业》。

  3. 创建Spark作业完成后，在控制台单击右上角“执行”提交作业，页面显示“批处理作业提交成功”说明Spark作业提交成功，可以在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”参数的说明。

完整示例代码

通过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

import org.apache.spark.sql.SparkSession;
 
public class java_rds {
 
    public static void main(String[] args) {
        SparkSession sparkSession = SparkSession.builder().appName("datasource-rds").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://192.168.6.150: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(3,'Liu',21),(4,'Joey',34)");
 
        //Read data from DLI data table
        sparkSession.sql("select * from dli_to_rds");
 
        //drop table
        sparkSession.sql("drop table dli_to_rds");
 
        sparkSession.close();
    }
}