On this page

Show all

Help Center/ Data Lake Insight/ Developer Guide/ Spark Jar Jobs/ Using Spark Jobs to Access Data Sources of Datasource Connections/ Connecting to CSS/ PySpark Example Code

PySpark Example Code

Updated on 2025-02-21 GMT+08:00

View PDF

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

Code implementation

Import dependency packages.

          
             from __future__ import print_function
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, Row
from pyspark.sql import SparkSession

Create a session.

          
             sparkSession = SparkSession.builder.appName("datasource-css").getOrCreate()

Connecting to data sources through DataFrame APIs

Set connection parameters.

          
             resource = "/mytest"
nodes = "to-css-1174404953-hDTx3UPK.datasource.com:9200"

NOTE:

resource indicates the name of the resource associated with the CSS. You can specify the resource location in /index/type format. (The index can be the database and type the table.)

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.

Create a schema and add data to it.

          
             schema = StructType([StructField("id", IntegerType(), False),                  
                     StructField("name", StringType(), False)])
rdd = sparkSession.sparkContext.parallelize([Row(1, "John"), Row(2, "Bob")])

Construct a DataFrame.

          
             dataFrame = sparkSession.createDataFrame(rdd, schema)

Save data to CSS.

          
             dataFrame.write.format("css").option("resource", resource).option("es.nodes", nodes).mode("Overwrite").save()

NOTE:

The options of mode 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.

Read data from CSS.

          
             jdbcDF = sparkSession.read.format("css").option("resource", resource).option("es.nodes", nodes).load()
jdbcDF.show()

View the operation result.

Connecting to data sources through SQL APIs

Create a table to connect to a CSS data source.

          
             sparkSession.sql(
    "create table css_table(id long, name string) using css options(  
    'es.nodes'='to-css-1174404953-hDTx3UPK.datasource.com:9200',
    'es.nodes.wan.only'='true',
    'resource'='/mytest')")

NOTE:

For details about the parameters for creating a CSS datasource connection table, see Table 1.

Insert data.

          
             sparkSession.sql("insert into css_table values(3,'tom')")

Query data.

          
             jdbcDF = sparkSession.sql("select * from css_table")
jdbcDF.show()

View the operation result.

Submitting a Spark job
1. Upload the Python code file to the OBS bucket.
2. In the Spark job editor, select the corresponding dependency module and execute the Spark job.
  NOTE:
  - For Spark 2.3.2 (soon to be take offline) or 2.4.5, set Module to sys.datasource.css when submitting a job.
  - If the Spark version is 3.1.1 or later, you do not need to select a module. Configure Spark parameters (--conf).
    spark.driver.extraClassPath=/usr/share/extension/dli/spark-jar/datasource/css/*
    
    spark.executor.extraClassPath=/usr/share/extension/dli/spark-jar/datasource/css/*
  - For how to submit a job on the console, see Table 3 "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

Connecting to data sources through DataFrame APIs

        
         
           
           # _*_ coding: utf-8 _*_
from __future__ import print_function
from pyspark.sql.types import Row, StructType, StructField, IntegerType, StringType
from pyspark.sql import SparkSession

if __name__ == "__main__":
  # Create a SparkSession session.   
  sparkSession = SparkSession.builder.appName("datasource-css").getOrCreate()

  # Setting cross-source connection parameters  
  resource = "/mytest"
  nodes = "to-css-1174404953-hDTx3UPK.datasource.com:9200"

  # Setting schema  
  schema = StructType([StructField("id", IntegerType(), False),       
                       StructField("name", StringType(), False)])

  # Construction data 
  rdd = sparkSession.sparkContext.parallelize([Row(1, "John"), Row(2, "Bob")])

  # Create a DataFrame from RDD and schema  
  dataFrame = sparkSession.createDataFrame(rdd, schema)

  # Write data to the CSS 
  dataFrame.write.format("css").option("resource", resource).option("es.nodes", nodes).mode("Overwrite").save()

  # Read data  
  jdbcDF = sparkSession.read.format("css").option("resource", resource).option("es.nodes", nodes).load()
  jdbcDF.show()

  # close session  
  sparkSession.stop()

          

        
       

Connecting to data sources through SQL APIs

        
         
           
           # _*_ coding: utf-8 _*_
from __future__ import print_function
from pyspark.sql import SparkSession

if __name__ == "__main__":
  # Create a SparkSession session.  
  sparkSession = SparkSession.builder.appName("datasource-css").getOrCreate()

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

  # Insert data into the DLI data table  
  sparkSession.sql("insert into css_table values(3,'tom')")

  # Read data from DLI data table   
  jdbcDF = sparkSession.sql("select * from css_table")   
  jdbcDF.show()

  # close session  
  sparkSession.stop()

          

        
       

CSS Security Cluster

Development description

Code implementation

Import dependency packages.

          
             from __future__ import print_function
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, Row
from pyspark.sql import SparkSession

Create a session and set the AKs and SKs.

NOTE:

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

          
             sparkSession = SparkSession.builder.appName("datasource-css").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 data sources through DataFrame APIs

Set connection parameters.

          
             resource = "/mytest";
nodes = "to-css-1174404953-hDTx3UPK.datasource.com:9200"

NOTE:

resource indicates the name of the resource associated with the CSS. You can specify the resource location in /index/type format. (The index can be the database and type the table.)

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.

Create a schema and add data to it.

          
             schema = StructType([StructField("id", IntegerType(), False),                  
                     StructField("name", StringType(), False)])
rdd = sparkSession.sparkContext.parallelize([Row(1, "John"), Row(2, "Bob")])

Construct a DataFrame.

          
             dataFrame = sparkSession.createDataFrame(rdd, schema)

Save data to CSS.

          
           
             
             dataFrame.write.format("css")
  .option("resource", resource)
  .option("es.nodes", nodes)
  .option("es.net.ssl", "true")
  .option("es.net.ssl.keystore.location", "obs://Bucket name/path/transport-keystore.jks")
  .option("es.net.ssl.keystore.pass", "***")
  .option("es.net.ssl.truststore.location", "obs://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("Overwrite")
  .save()

            

          
         

NOTE:

The options of mode 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.

Read data from CSS.

          
           
             
             jdbcDF = sparkSession.read.format("css")\
  .option("resource", resource)\
  .option("es.nodes", nodes)\
  .option("es.net.ssl", "true")\
  .option("es.net.ssl.keystore.location", "obs://Bucket name/path/transport-keystore.jks")\
  .option("es.net.ssl.keystore.pass", "***")\
  .option("es.net.ssl.truststore.location", "obs://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()
jdbcDF.show()

            

          
         

View the operation result.

Connecting to data sources through SQL APIs

Create a table to connect to a CSS data source.

          
           
             
             sparkSession.sql(
        "create table css_table(id long, name string) using css options(\  
        'es.nodes'='to-css-1174404953-hDTx3UPK.datasource.com:9200',\
        'es.nodes.wan.only'='true',\
        'resource'='/mytest',\
	'es.net.ssl'='true',\
	'es.net.ssl.keystore.location'='obs://Bucket name/path/transport-keystore.jks',\
	'es.net.ssl.keystore.pass'='***',\
	'es.net.ssl.truststore.location'='obs://Bucket name/path/truststore.jks',\
	'es.net.ssl.truststore.pass'='***',\
	'es.net.http.auth.user'='admin',\
	'es.net.http.auth.pass'='***')")

            

          
         

NOTE:

For details about the parameters for creating a CSS datasource connection table, see Table 1.

Insert data.

          
             sparkSession.sql("insert into css_table values(3,'tom')")

Query data.

          
             jdbcDF = sparkSession.sql("select * from css_table")
jdbcDF.show()

View the operation result.

Submitting a Spark job
1. Upload the Python code file to DLI. For details about console operations, see Creating a Package. For 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 API operations, see Creating a Batch Processing Job.
  NOTE:
  - When submitting a job, you need to specify a dependency module named sys.datasource.css.
  - For how to submit a job on the console, see Table 3 "Parameters for selecting dependency resources" in Creating a Spark Job.
  - For details about how to submit a job through an API, see the modules parameter in Request parameters of Creating a Batch Processing Job in the Data Lake Insight API Reference.

Complete example code

Connecting to data sources through DataFrame APIs

NOTE:

        
         
           
           # _*_ coding: utf-8 _*_
from __future__ import print_function
from pyspark.sql.types import Row, StructType, StructField, IntegerType, StringType
from pyspark.sql import SparkSession

if __name__ == "__main__":
  # Create a SparkSession session.   
  sparkSession = SparkSession.builder.appName("datasource-css").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")

  # Setting cross-source connection parameters  
  resource = "/mytest";
  nodes = "to-css-1174404953-hDTx3UPK.datasource.com:9200"

  # Setting schema  
  schema = StructType([StructField("id", IntegerType(), False),       
                       StructField("name", StringType(), False)])

  # Construction data 
  rdd = sparkSession.sparkContext.parallelize([Row(1, "John"), Row(2, "Bob")])

  # Create a DataFrame from RDD and schema  
  dataFrame = sparkSession.createDataFrame(rdd, schema)

  # Write data to the CSS 
  dataFrame.write.format("css")
    .option("resource", resource)
    .option("es.nodes", nodes)
    .option("es.net.ssl", "true")
    .option("es.net.ssl.keystore.location", "obs://Bucket name/path/transport-keystore.jks")
    .option("es.net.ssl.keystore.pass", "***")
    .option("es.net.ssl.truststore.location", "obs://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("Overwrite")
    .save()

  # Read data  
  jdbcDF = sparkSession.read.format("css")\
    .option("resource", resource)\
    .option("es.nodes", nodes)\
    .option("es.net.ssl", "true")\
    .option("es.net.ssl.keystore.location", "obs://Bucket name/path/transport-keystore.jks")\
    .option("es.net.ssl.keystore.pass", "***")\
    .option("es.net.ssl.truststore.location", "obs://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()
  jdbcDF.show()

  # close session  
  sparkSession.stop()

          

        
       

Connecting to data sources through SQL APIs

        
         
           
           # _*_ coding: utf-8 _*_
from __future__ import print_function
from pyspark.sql import SparkSession
import os               
 
if __name__ == "__main__":
  
  # Create a SparkSession session.   
  sparkSession = SparkSession.builder.appName("datasource-css").getOrCreate()
  # Create a DLI data table for DLI-associated CSS   
  sparkSession.sql("create table css_table(id int, name string) using css options(\
                    'es.nodes'='192.168.6.204:9200',\
                    'es.nodes.wan.only'='true',\
                    'resource'='/mytest',\
                    'es.net.ssl'='true',\
                    'es.net.ssl.keystore.location' = 'obs://xietest1/lzq/keystore.jks',\
                    'es.net.ssl.keystore.pass' = '**',\
                    'es.net.ssl.truststore.location'='obs://xietest1/lzq/truststore.jks',\
                    'es.net.ssl.truststore.pass'='**',\
                    'es.net.http.auth.user'='admin',\
                    'es.net.http.auth.pass'='**')")
 
  # Insert data into the DLI data table  
  sparkSession.sql("insert into css_table values(3,'tom')")
 
  # Read data from DLI data table   
  jdbcDF = sparkSession.sql("select * from css_table")   
  jdbcDF.show()
 
  # close session  
  sparkSession.stop()

          

        
       

Parent topic: Connecting to CSS

Previous topic: Scala Example Code

Next topic: Java Example Code

Feedback

Was this page helpful?

Helpful Not helpful

Provide feedback

Thank you very much for your feedback. We will continue working to improve the documentation.See the reply and handling status in My Cloud VOC.

The system is busy. Please try again later.

Which of the following issues have you encountered?

Content is inconsistent with the product UI

Unclear descriptions

Lack of examples or code

Incorrect steps

Can't find what I need

Lack of best practices

Feedback (optional)

0/500

Select at least one type of issue, and enter your comments or suggestions.

Enter a maximum of 500 characters.

Submit Cancel

For any further questions, feel free to contact us through the chatbot.

Chatbot