更新时间:2024-03-06 GMT+08:00
分享

java样例代码

开发说明

本样例只适用于MRS的OpenTSDB。

  • 前提条件

    在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.builder().appName("datasource-opentsdb").getOrCreate();
      
  • 通过SQL API 访问
    • 创建DLI跨源访问MRS OpenTSDB的关联表,填写连接参数。
      1
      sparkSession.sql("create table opentsdb_new_test using opentsdb options('Host'='10.0.0.171:4242','metric'='ctopentsdb','tags'='city,location')");
      

      Host、metric、tags三个参数详情讲解可参考表1

    • 插入数据
      1
      sparkSession.sql("insert into opentsdb_new_test values('Penglai', 'abc', '2021-06-30 18:00:00', 30.0)");
      
    • 查询数据
      1
      sparkSession.sql("select * from opentsdb_new_test").show();
      

      插入数据后:

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

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

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

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

      • 如果选择spark版本为2.3.2(即将下线)或2.4.5提交作业时,需要指定Module模块,名称为:sys.datasource.opentsdb。
      • 如果选择Spark版本为3.1.1时,无需选择Module模块, 需在 'Spark参数(--conf)' 配置

        spark.driver.extraClassPath=/usr/share/extension/dli/spark-jar/datasource/opentsdb/*

        spark.executor.extraClassPath=/usr/share/extension/dli/spark-jar/datasource/opentsdb/*

      • 通过控制台提交作业请参考《数据湖探索用户指南》中的“选择依赖资源参数说明”表说明
      • 通过API提交作业请参考《数据湖探索API参考》>《创建批处理作业》中“表2-请求参数说明”关于“modules”参数的说明。

完整示例代码

  • Maven依赖
    <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
    import org.apache.spark.sql.SparkSession;
     
    public class java_mrs_opentsdb {
     
        private static SparkSession sparkSession = null;
     
        public static void main(String[] args) {
            //create a SparkSession session
            sparkSession = SparkSession.builder().appName("datasource-opentsdb").getOrCreate();
     
            sparkSession.sql("create table opentsdb_new_test using opentsdb options('Host'='10.0.0.171:4242','metric'='ctopentsdb','tags'='city,location')");
     
            //*****************************SQL module***********************************
            sparkSession.sql("insert into opentsdb_new_test values('Penglai', 'abc', '2021-06-30 18:00:00', 30.0)");
            System.out.println("Penglai new timestamp");
            sparkSession.sql("select * from opentsdb_new_test").show();
     
           sparkSession.close();
     
        }
    }
    
分享:

    相关文档

    相关产品