Scala Example Code

Development Description

The CloudTable OpenTSDB and MRS OpenTSDB can be connected to DLI as data sources.

• Prerequisites

  A datasource connection has been created on the DLI management console. For details, see Enhanced Datasource Connections.

  

  Hard-coded or plaintext passwords pose significant security risks. To ensure security, encrypt your passwords, store them in configuration files or environment variables, and decrypt them when needed.

• Constructing dependency information and creating a Spark session
  1. Import dependencies.
     Maven dependency involved

     1 2 3 4 5

     <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.11</artifactId> <version>2.3.2</version> </dependency>

     Import dependency packages.

     1 2 3 4

     import scala.collection.mutable import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.rdd.RDD import org.apache.spark.sql.types._

  2. Create a session.

     1	val sparkSession = SparkSession.builder().getOrCreate()

  3. Create a table to connect to an OpenTSDB data source.

     1 2 3 4

     sparkSession.sql("create table opentsdb_test using opentsdb options( 'Host'='opentsdb-3xcl8dir15m58z3.cloudtable.com:4242', 'metric'='ctopentsdb', 'tags'='city,location')")

     Table 1 Parameters for creating a table

     Parameter	Description
     host	OpenTSDB IP address. To access CloudTable OpenTSDB, specify the OpenTSDB connection address. You can log in to the CloudTable console, choose Cluster Mode and click the target cluster name, and obtain the OpenTSDB connection address from the cluster information. You can also access the MRS OpenTSDB. If you have created an enhanced datasource connection, enter the IP address and port number of the node where the OpenTSDB is located. The format is IP:PORT. If the OpenTSDB has multiple nodes, separate their IP addresses by semicolons (;). For details about how to obtain the IP address, see MRS cluster OpenTSDB IP address and MRS cluster OpenTSDB port number. If you use a basic datasource connection, enter the connection address returned. For details about operations on the management console, see the Data Lake Insight User Guide.
     metric	Name of the metric in OpenTSDB corresponding to the DLI table to be created.
     tags	Tags corresponding to the metric, used for operations such as classification, filtering, and quick search. A maximum of 8 tags, including all tagk values under the metric, can be added and are separated by commas (,).

• Connecting to data sources through SQL APIs
  1. Insert data.

     1	sparkSession.sql("insert into opentsdb_test values('futian', 'abc', '1970-01-02 18:17:36', 30.0)")

  2. Query data.

     1	sparkSession.sql("select * from opentsdb_test").show()

     Response

     

• Connecting to data sources through DataFrame APIs
  1. Construct a schema.

     1 2 3 4 5

     val attrTag1Location = new StructField("location", StringType) val attrTag2Name = new StructField("name", StringType) val attrTimestamp = new StructField("timestamp", LongType) val attrValue = new StructField("value", DoubleType) val attrs = Array(attrTag1Location, attrTag2Name, attrTimestamp, attrValue)

  2. Construct data based on the schema type.

     1 2	val mutableRow: Seq[Any] = Seq("aaa", "abc", 123456L, 30.0) val rddData: RDD[Row] = sparkSession.sparkContext.parallelize(Array(Row.fromSeq(mutableRow)), 1)

  3. Import data to OpenTSDB.

     1	sparkSession.createDataFrame(rddData, new StructType(attrs)).write.insertInto("opentsdb_test")

  4. Read data from OpenTSDB.

     1 2 3 4 5

     val map = new mutable.HashMap[String, String]() map("metric") = "ctopentsdb" map("tags") = "city,location" map("Host") = "opentsdb-3xcl8dir15m58z3.cloudtable.com:4242" sparkSession.read.format("opentsdb").options(map.toMap).load().show()

     Response

     

• Submitting a Spark job
  1. Generate a JAR package based on the code and upload the package to DLI.

     For details about console operations, see Creating a Package. For details about API operations, see Uploading a Package Group.

  2. In the Spark job editor, select the corresponding dependency module and execute the Spark job.
     For details about console operations, see Creating a Spark Job. For details about API operations, see Creating a Batch Processing Job.

     

     • If the Spark version is 2.3.2 (will be offline soon) or 2.4.5, specify the Module to sys.datasource.opentsdb when you submit a job.
     • If the Spark version is 3.1.1, you do not need to select a module. Configure Spark parameters (--conf).

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

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

     • For details about how to submit a job on the console, see the description of the table "Parameters for selecting dependency resources" in Creating a Spark Job.
     • For details about how to submit a job through an API, see the description of the modules parameter in Table 2 "Request parameters" in Creating a Batch Processing Job.

Complete Example Code

• Maven dependency

  1 2 3 4 5

  <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.11</artifactId> <version>2.3.2</version> </dependency>

• Connecting to data sources through SQL APIs

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20

  import org.apache.spark.sql.SparkSession object Test_OpenTSDB_CT { def main(args: Array[String]): Unit = { // Create a SparkSession session. val sparkSession = SparkSession.builder().getOrCreate() // Create a data table for DLI association OpenTSDB sparkSession.sql("create table opentsdb_test using opentsdb options( 'Host'='opentsdb-3xcl8dir15m58z3.cloudtable.com:4242', 'metric'='ctopentsdb', 'tags'='city,location')") //*****************************SQL module*********************************** sparkSession.sql("insert into opentsdb_test values('futian', 'abc', '1970-01-02 18:17:36', 30.0)") sparkSession.sql("select * from opentsdb_test").show() sparkSession.close() } }

• Connecting to data sources through DataFrame APIs

  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

  import scala.collection.mutable import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.rdd.RDD import org.apache.spark.sql.types._ object Test_OpenTSDB_CT { def main(args: Array[String]): Unit = { // Create a SparkSession session. val sparkSession = SparkSession.builder().getOrCreate() // Create a data table for DLI association OpenTSDB sparkSession.sql("create table opentsdb_test using opentsdb options( 'Host'='opentsdb-3xcl8dir15m58z3.cloudtable.com:4242', 'metric'='ctopentsdb', 'tags'='city,location')") //*****************************DataFrame model*********************************** // Setting schema val attrTag1Location = new StructField("location", StringType) val attrTag2Name = new StructField("name", StringType) val attrTimestamp = new StructField("timestamp", LongType) val attrValue = new StructField("value", DoubleType) val attrs = Array(attrTag1Location, attrTag2Name, attrTimestamp,attrValue) // Populate data according to the type of schema val mutableRow: Seq[Any] = Seq("aaa", "abc", 123456L, 30.0) val rddData: RDD[Row] = sparkSession.sparkContext.parallelize(Array(Row.fromSeq(mutableRow)), 1) //Import the constructed data into OpenTSDB sparkSession.createDataFrame(rddData, new StructType(attrs)).write.insertInto("opentsdb_test") //Read data on OpenTSDB val map = new mutable.HashMap[String, String]() map("metric") = "ctopentsdb" map("tags") = "city,location" map("Host") = "opentsdb-3xcl8dir15m58z3.cloudtable.com:4242" sparkSession.read.format("opentsdb").options(map.toMap).load().show() sparkSession.close() } }