开发说明

本样例只适用于MRS的OpenTSDB。

• 前提条件

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

• 代码实现
  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参考》>《创建批处理作业》。
     

     • 提交作业时，需要指定Module模块，名称为：sys.datasource.opentsdb。
     • 通过控制台提交作业请参考《数据湖探索用户指南》中的“表6-选择依赖资源参数说明”。
     • 通过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(); } }