# pyspark样例代码
#### 开发说明
支持对接CloudTable的OpenTSDB和MRS的OpenTSDB。
- 前提条件 在DLI管理控制台上已完成创建跨源连接。具体操作请参考《[数据湖探索用户指南](https://support.huaweicloud.com/usermanual-dli/dli_01_0426.html)》。
  ![](https://support.huaweicloud.com/devg-dli/public_sys-resources/note_3.0-zh-cn.png)
  认证用的password硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。
  
- 代码实现详解
  1. import相关依赖包
     ```
     from __future__ import print_function
     from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType
     from pyspark.sql import SparkSession
     ```
     
  
  2. 创建会话
     ```
     sparkSession = SparkSession.builder.appName("datasource-opentsdb").getOrCreate()
     ```
     
  
  3. 创建DLI跨源访问 OpenTSDB的关联表
     ```
     sparkSession.sql("create table opentsdb_test using opentsdb options(
       'Host'='opentsdb-3xcl8dir15m58z3.cloudtable.com:4242',
       'metric'='ct_opentsdb',
       'tags'='city,location')")
     ```
     ![](https://support.huaweicloud.com/devg-dli/public_sys-resources/note_3.0-zh-cn.png)
     Host、metric、tags三个参数详情讲解可参考[表1](https://support.huaweicloud.com/devg-dli/dli_09_0065.html#dli_09_0065__zh-cn_topic_0190597601_table463015581831)。
     
   
- 通过SQL API访问
  1. 插入数据
     ```
     sparkSession.sql("insert into opentsdb_test values('aaa', 'abc', '2021-06-30 18:00:00', 30.0)")
     ```
     
  
  2. 查询数据
     ```
     result = sparkSession.sql("SELECT * FROM opentsdb_test")
     ```
     
   
- 通过DataFrame API 访问
  1. 构造schema
     ```
     schema = StructType([StructField("location", StringType()),\                     
                          StructField("name", StringType()), \                   
                          StructField("timestamp", LongType()),\                  
                          StructField("value", DoubleType())])
     ```
     
  
  2. 设置数据
     ```
     dataList = sparkSession.sparkContext.parallelize([("aaa", "abc", 123456L, 30.0)])
     ```
     
  
  3. 创建DataFrame
     ```
     dataFrame = sparkSession.createDataFrame(dataList, schema)
     ```
     
  
  4. 导入数据到OpenTSDB
     ```
     dataFrame.write.insertInto("opentsdb_test")
     ```
     
  
  5. 读取OpenTSDB上的数据
     ```
     jdbdDF = sparkSession.read
         .format("opentsdb")\
         .option("Host","opentsdb-3xcl8dir15m58z3.cloudtable.com:4242")\
         .option("metric","ctopentsdb")\
         .option("tags","city,location")\
         .load()
     jdbdDF.show()
     ```
     
  
  6. 操作结果 ![](https://support.huaweicloud.com/devg-dli/zh-cn_image_0223997306.png)
     
   
- 提交Spark作业
  1. 将写好的python代码文件上传至OBS桶中。
  
  2. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。
     ![](https://support.huaweicloud.com/devg-dli/public_sys-resources/note_3.0-zh-cn.png)
     - 如果选择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/\*
       
     
     - 通过控制台提交作业请参考《[数据湖探索用户指南](https://support.huaweicloud.com/usermanual-dli/dli_01_0384.html)》中的"选择依赖资源参数说明"。
     
     - 通过API提交作业请参考《数据湖探索API参考》\>《[创建批处理作业](https://support.huaweicloud.com/api-dli/dli_02_0124.html)》中"表2-请求参数说明"关于"modules"参数的说明。
       
   
 
#### 完整示例代码
- 通过SQL API访问MRS的OpenTSDB
  ```
  # _*_ coding: utf-8 _*_
  from __future__ import print_function
  from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType
  from pyspark.sql import SparkSession
  if __name__ == "__main__":
    # Create a SparkSession session.    
    sparkSession = SparkSession.builder.appName("datasource-opentsdb").getOrCreate()
    # Create a DLI cross-source association opentsdb data table    
    sparkSession.sql(\
      "create table opentsdb_test using opentsdb options(\
      'Host'='10.0.0.171:4242',\
      'metric'='cts_opentsdb',\
      'tags'='city,location')")
    sparkSession.sql("insert into opentsdb_test values('aaa', 'abc', '2021-06-30 18:00:00', 30.0)")
    result = sparkSession.sql("SELECT * FROM opentsdb_test")
    result.show()
    # close session 
    sparkSession.stop()
  ```
  
- 通过DataFrame API访问OpenTSDB
  ```
  # _*_ coding: utf-8 _*_
  from __future__ import print_function
  from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType
  from pyspark.sql import SparkSession
  if __name__ == "__main__":
    # Create a SparkSession session.    
    sparkSession = SparkSession.builder.appName("datasource-opentsdb").getOrCreate()
    # Create a DLI cross-source association opentsdb data table    
    sparkSession.sql(
      "create table opentsdb_test using opentsdb options(\
      'Host'='opentsdb-3xcl8dir15m58z3.cloudtable.com:4242',\
      'metric'='ct_opentsdb',\
      'tags'='city,location')")
    # Create a DataFrame and initialize the DataFrame data.    
    dataList = sparkSession.sparkContext.parallelize([("aaa", "abc", 123456L, 30.0)])
    # Setting schema   
    schema = StructType([StructField("location", StringType()),\                 
                         StructField("name", StringType()),\                
                         StructField("timestamp", LongType()),\               
                         StructField("value", DoubleType())])
    # Create a DataFrame from RDD and schema   
    dataFrame = sparkSession.createDataFrame(dataList, schema)
    # Set cross-source connection parameters   
    metric = "ctopentsdb"   
    tags = "city,location"  
    Host = "opentsdb-3xcl8dir15m58z3.cloudtable.com:4242"
    # Write data to the cloudtable-opentsdb   
    dataFrame.write.insertInto("opentsdb_test")
    # ******* Opentsdb does not currently implement the ctas method to save data, so the save() method cannot be used.*******   
    # dataFrame.write.format("opentsdb").option("Host", Host).option("metric", metric).option("tags", tags).mode("Overwrite").save()   
    # Read data on CloudTable-OpenTSDB  
    jdbdDF = sparkSession.read\       
        .format("opentsdb")\      
        .option("Host",Host)\     
        .option("metric",metric)\  
        .option("tags",tags)\    
        .load()   
    jdbdDF.show()
    # close session 
    sparkSession.stop()
  ```