Help Center> Data Lake Insight> Developer Guide> Using the Spark Job to Access Data Sources of Datasource Connections> Connecting to CSS> Scala Example Code
View PDF

Scala Example Code

Prerequisites

A datasource connection has been created on the DLI management console. For details, see Data Lake Insight User Guide.

CSS Non-security Cluster

  • Development description
    • Construct dependency information and create a Spark session.
      1. Import dependencies
        Involved 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>
        Dependencies related to import 
        1
2
        		 
        import org.apache.spark.sql.{Row, SaveMode, SparkSession}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
      2. Create a session 
        1
        		 
        val sparkSession = SparkSession.builder().getOrCreate()
    • Connecting to datasources through SQL APIs
      1. Create a table to connect to CSS datasource 
        1
2
3
4
        		 
        sparkSession.sql("create table css_table(id int, name string) using css options(
	'nodes' 'to-css-1174404221-Y2bKVIqY.datasource.com:9200',
	'nodes.wan.only'='true',
	'resource' '/mytest/css')")
        Table 1 Parameters for creating a table

        Parameter

        Description

        es.nodes

        CSS connection address. You need to create a datasource connection first. For details, see Data Lake Insight User Guide.

        After a basic datasource connection is created, the returned IP address is used.

        After an enhanced datasource connection is created, use the intranet IP address provided by CSS. The address format is IP1:PORT1,IP2:PORT2.

        resource

        The resource is used to specify the CSS datasource connection name. You can use /index/type to specify the resource location (for easier understanding, the index can be seen as database and type as table).

        NOTE:
        • In Elasticsearch 6.X, a single index supports only one type, and the type name can be customized.
        • In Elasticsearch 7.X, a single index uses _doc as the type name and cannot be customized. To access Elasticsearch 7.X, set this parameter to index.

        pushdown

        Indicates whether the press function of CSS is enabled. The default value is set to true. If there are a large number of I/O transfer tables, the pushdown can be enabled to reduce I/Os when the where filtering conditions are met.

        strict

        Indicates whether the CSS pushdown is strict. The default value is set to false. In exact match scenarios, more I/Os are reduced than pushdown.

        batch.size.entries

        Maximum number of entries that can be inserted to a batch processing. The default value is 1000. If the size of a single data record is so large that the number of data records in the bulk storage reaches the upper limit of the data amount of a single batch processing, the system stops storing data and submits the data based on the batch.size.bytes.

        batch.size.bytes

        Maximum amount of data in a single batch processing. The default value is 1 MB. If the size of a single data record is so small that the number of data records in the bulk storage reaches the upper limit of the data amount of a single batch processing, the system stops storing data and submits the data based on the batch.size.entries.

        es.nodes.wan.only

        Indicates whether to access the Elasticsearch node using only the domain name. The default value is false. If a basic datasource connection address is used as the es.nodes, set this parameter to true. If the original internal IP address provided by CSS is used as the es.nodes, you do not need to set this parameter or set it to false.

        es.mapping.id

        Specifies a field whose value is used as the document ID in the Elasticsearch node.

        NOTE:
        • The document ID in the same /index/type is unique. If a field that functions as a document ID has duplicate values, the document with the duplicate ID will be overwritten when the ES is inserted.
        • This feature can be used as a fault tolerance solution. When data is being inserted, the DLI job fails and some data has been inserted into Elasticsearch. The data is redundant. If Document id is set, the last redundant data will be overwritten when the DLI job is executed again.

        batch.size.entries and batch.size.bytes limit the number of data records and data volume respectively.

      2. Insert data 
        1
        		 
        sparkSession.sql("insert into css_table values(13, 'John'),(22, 'Bob')")
      3. Query data 
        1
2
        		 
        val dataFrame = sparkSession.sql("select * from css_table")
dataFrame.show()

        Before data is inserted:

        After data is inserted:

      4. Delete the datasource connection table 
        1
        		 
        sparkSession.sql("drop table css_table")
    • Connecting to datasources through DataFrame APIs
      1. Configure datasource connection 
        1
2
        		 
        val resource = "/mytest/css"
val nodes = "to-css-1174405013-Ht7O1tYf.datasource.com:9200"
      2. Create a schema and add data to the schema 
        1
2
        		 
        val schema = StructType(Seq(StructField("id", IntegerType, false), StructField("name", StringType, false)))
val rdd = sparkSession.sparkContext.parallelize(Seq(Row(12, "John"),Row(21,"Bob")))
      3. Import data to CSS 
        1
2
3
4
5
6
7
        		 
        val dataFrame_1 = sparkSession.createDataFrame(rdd, schema)
dataFrame_1.write 
  .format("css") 
  .option("resource", resource) 
  .option("nodes", nodes) 
  .mode(SaveMode.Append) 
  .save()

        The value of SaveMode can be one of the following:

        • ErrorIfExis: If the data already exists, the system throws an exception.
        • Overwrite: If the data already exists, the original data will be overwritten.
        • Append: If the data already exists, the system saves the new data.
        • Ignore: If the data already exists, no operation is required. This is similar to the SQL statement CREATE TABLE IF NOT EXISTS.
      4. Read data from CSS 
        1
2
        		 
        val dataFrameR = sparkSession.read.format("css").option("resource",resource).option("nodes", nodes).load()
dataFrameR.show()

        Before data is inserted:

        After data is inserted:

    • 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 the Data Lake Insight User Guide. For API references, see Uploading a Resource Package in the Data Lake Insight API Reference.
      2. In the Spark job editor, select the corresponding dependency and execute the Spark job. For details about console operations, see the Data Lake Insight User Guide. For API references, see Creating a Batch Processing Job in the Data Lake Insight API Reference.
        • When submitting a job, you need to specify a dependency module named sys.datasource.css.
        • For details about how to submit a job on the console, see Table 6-Dependency Resources parameter description in the Data Lake Insight User Guide.
        • For details about how to submit a job through an API, see the modules parameter in Table 2-Request parameter description of Creating a Batch Processing Job in the Data Lake Insight API Reference.
  • 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 datasources through SQL 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
      		 
      import org.apache.spark.sql.SparkSession

object Test_SQL_CSS {
  def main(args: Array[String]): Unit = {
    // Create a SparkSession session.
    val sparkSession = SparkSession.builder().getOrCreate()

    // Create a DLI data table for DLI-associated CSS
    sparkSession.sql("create table css_table(id long, name string) using css options(
	'nodes' = 'to-css-1174404217-QG2SwbVV.datasource.com:9200',
	'nodes.wan.only' = 'true',
	'resource' = '/mytest/css')")

    //*****************************SQL model***********************************
    // Insert data into the DLI data table
    sparkSession.sql("insert into css_table values(13, 'John'),(22, 'Bob')")
   
    // Read data from DLI data table
    val dataFrame = sparkSession.sql("select * from css_table")
    dataFrame.show()
   
    // drop table
    sparkSession.sql("drop table css_table")

    sparkSession.close()
  }
}
    • Connecting to datasources 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
      		 
      import org.apache.spark.sql.{Row, SaveMode, SparkSession};
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType};

object Test_SQL_CSS {
  def main(args: Array[String]): Unit = {
    //Create a SparkSession session.
    val sparkSession = SparkSession.builder().getOrCreate()

    //*****************************DataFrame model***********************************
    // Setting the /index/type of CSS
    val resource = "/mytest/css"
  
    // Define the cross-origin connection address of the CSS cluster
    val nodes = "to-css-1174405013-Ht7O1tYf.datasource.com:9200"

    //Setting schema
    val schema = StructType(Seq(StructField("id", IntegerType, false), StructField("name", StringType, false)))
  
    // Construction data
    val rdd = sparkSession.sparkContext.parallelize(Seq(Row(12, "John"),Row(21,"Bob")))
  
    // Create a DataFrame from RDD and schema
    val dataFrame_1 = sparkSession.createDataFrame(rdd, schema)
  
   //Write data to the CSS
   dataFrame_1.write .format("css") 
    .option("resource", resource) 
    .option("nodes", nodes) 
    .mode(SaveMode.Append) 
    .save()
  
    //Read data
    val dataFrameR = sparkSession.read.format("css").option("resource", resource).option("nodes", nodes).load()
    dataFrameR.show()

    spardSession.close()
  }
}

CSS Security Cluster

  • Preparations
    • The Elasticsearch 6.5.4 provided by CSS provides the security settings. Once the function is enabled, CSS provides identity authentication, authorization, and encryption for users. Before connecting DLI to the CSS security cluster, you need to perform certain preparations.
      1. Select CSS Elasticsearch 6.5.4 or a later cluster version, create a CSS security cluster, and download the security cluster certificate (CloudSearchService.cer).
      2. Create a datasource connection.
      3. Use the keytools tool to generate the keystore and truststore files.
        1. While generating the required files, the security certificate (CloudSearchService.cer) of the security cluster is required. Run the following commands to generate the files. Other parameters of the keytools tool can be set as required. 
          keytool -genkeypair -alias certificatekey -keyalg RSA -keystore transport-keystore.jks
keytool -list -v -keystore transport-keystore.jks
keytool -import -alias certificatekey -file CloudSearchService.cer  -keystore truststore.jks
keytool -list -v -keystore truststore
        2. Upload the generated keystore and truststore files to an OBS bucket.
  • Development description
    • Construct dependency information and create a Spark session.
      1. Import dependencies
        Involved 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>
        Dependencies related to import 
        1
2
        		 
        import org.apache.spark.sql.{Row, SaveMode, SparkSession}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
      2. Create a session and set the access keys. 
        1
2
3
4
5
        		 
        val sparkSession = SparkSession.builder().getOrCreate()
sparkSession.conf.set("fs.obs.access.key", ak)
sparkSession.conf.set("fs.obs.secret.key", sk)
sparkSession.conf.set("fs.obs.endpoint", enpoint)
sparkSession.conf.set("fs.obs.connecton.ssl.enabled", "false")
    • Connecting to datasources through DataFrame APIs
      1. Configure datasource connection 
        1
2
        		 
        val resource = "/mytest/css"
val nodes = "to-css-1174405013-Ht7O1tYf.datasource.com:9200"
      2. Create a schema and add data to the schema 
        1
2
        		 
        val schema = StructType(Seq(StructField("id", IntegerType, false), StructField("name", StringType, false)))
val rdd = sparkSession.sparkContext.parallelize(Seq(Row(12, "John"),Row(21,"Bob")))
      3. Import data to CSS 
         1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
        		 
        val dataFrame_1 = sparkSession.createDataFrame(rdd, schema)
dataFrame_1.write 
  .format("css") 
  .option("resource", resource) 
  .option("nodes", nodes) 
  .option("es.net.ssl", "true")
  .option("es.net.ssl.keystore.location", "obs://AK:SK@bucket name/path/transport-keystore.jks")
  .option("es.net.ssl.keystore.pass", "***")
  .option("es.net.ssl.truststore.location", "obs://AK:SK@bucket name/path/truststore.jks")
  .option("es.net.ssl.truststore.pass", "***")
  .option("es.net.http.auth.user", "admin")
  .option("es.net.http.auth.pass", "***")
  .mode(SaveMode.Append) 
  .save()

        The value of SaveMode can be one of the following:

        • ErrorIfExis: If the data already exists, the system throws an exception.
        • Overwrite: If the data already exists, the original data will be overwritten.
        • Append: If the data already exists, the system saves the new data.
        • Ignore: If the data already exists, no operation is required. This is similar to the SQL statement CREATE TABLE IF NOT EXISTS.
      4. Read data from CSS 
         1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
        		 
        val dataFrameR = sparkSession.read.format("css")
        .option("resource",resource)
        .option("nodes", nodes)
        .option("es.net.ssl", "true")
        .option("es.net.ssl.keystore.location", "obs://AK:SK@bucket name/path/transport-keystore.jks")
        .option("es.net.ssl.keystore.pass", "***")
        .option("es.net.ssl.truststore.location", "obs://AK:SK@bucket name/path/truststore.jks")
        .option("es.net.ssl.truststore.pass", "***")
        .option("es.net.http.auth.user", "admin")
        .option("es.net.http.auth.pass", "***")
        .load()
dataFrameR.show()

        Before data is inserted:

        After data is inserted:

    • 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 the Data Lake Insight User Guide. For API references, see Uploading a Resource Package in the Data Lake Insight API Reference.
      2. In the Spark job editor, select the corresponding dependency and execute the Spark job. For details about console operations, see the Data Lake Insight User Guide. For API references, see Creating a Batch Processing Job in the Data Lake Insight API Reference.
        • When submitting a job, you need to specify a dependency module named sys.datasource.css.
        • For details about how to submit a job on the console, see Table 6-Dependency Resources parameter description in the Data Lake Insight User Guide.
        • For details about how to submit a job through an API, see the modules parameter in Table 2-Request parameter description of Creating a Batch Processing Job in the Data Lake Insight API Reference.
  • 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 datasources 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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
      		 
      import org.apache.spark.sql.{Row, SaveMode, SparkSession};
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType};

object Test_SQL_CSS {
  def main(args: Array[String]): Unit = {
    //Create a SparkSession session.
    val sparkSession = SparkSession.builder().getOrCreate()
    sparkSession.conf.set("fs.obs.access.key", ak)
    sparkSession.conf.set("fs.obs.secret.key", sk)

    //*****************************DataFrame model***********************************
    // Setting the /index/type of CSS
    val resource = "/mytest/css"
  
    // Define the cross-origin connection address of the CSS cluster
    val nodes = "to-css-1174405013-Ht7O1tYf.datasource.com:9200"

    //Setting schema
    val schema = StructType(Seq(StructField("id", IntegerType, false), StructField("name", StringType, false)))
  
    // Construction data
    val rdd = sparkSession.sparkContext.parallelize(Seq(Row(12, "John"),Row(21,"Bob")))
  
    // Create a DataFrame from RDD and schema
    val dataFrame_1 = sparkSession.createDataFrame(rdd, schema)
  
   //Write data to the CSS
   dataFrame_1.write .format("css") 
    .option("resource", resource) 
    .option("nodes", nodes) 
    .option("es.net.ssl", "true")
    .option("es.net.ssl.keystore.location", "obs://AK:SK@bucket name/path/transport-keystore.jks")
    .option("es.net.ssl.keystore.pass", "***")
    .option("es.net.ssl.truststore.location", "obs://AK:SK@bucket name/path/truststore.jks")
    .option("es.net.ssl.truststore.pass", "***")
    .option("es.net.http.auth.user", "admin")
    .option("es.net.http.auth.pass", "***")
    .mode(SaveMode.Append) 
    .save();
  
    //Read data
    val dataFrameR = sparkSession.read.format("css")
    .option("resource", resource)
    .option("nodes", nodes)
    .option("es.net.ssl", "true")
    .option("es.net.ssl.keystore.location", "obs://AK:SK@bucket name/path/transport-keystore.jks")
    .option("es.net.ssl.keystore.pass", "***")
    .option("es.net.ssl.truststore.location", "obs://AK:SK@bucket name/path/truststore.jks")
    .option("es.net.ssl.truststore.pass", "***")
    .option("es.net.http.auth.user", "admin")
    .option("es.net.http.auth.pass", "***")
    .load()
    dataFrameR.show()

    spardSession.close()
  }
}
Parent topic: Connecting to CSS