更新时间:2022-05-09 GMT+08:00
分享

pyspark样例代码

开发说明

支持对接CloudTable的OpenTSDB和MRS的OpenTSDB。

  • 前提条件

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

  • 代码实现详解
    1. import相关依赖包
      1
      2
      3
      from __future__ import print_function
      from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType
      from pyspark.sql import SparkSession
      
    2. 创建会话
      1
      sparkSession = SparkSession.builder.appName("datasource-opentsdb").getOrCreate()
      
    3. 创建DLI跨源访问 OpenTSDB的关联表
      1
      2
      3
      4
      sparkSession.sql("create table opentsdb_test using opentsdb options(
        'Host'='opentsdb-3xcl8dir15m58z3.cloudtable.com:4242',
        'metric'='ct_opentsdb',
        'tags'='city,location')")
      

      Host、metric、tags三个参数详情讲解可参考表1

  • 通过SQL API访问
    1. 插入数据
      sparkSession.sql("insert into opentsdb_test values('beijing', 'abc', '2021-06-30 18:00:00', 30.0)")
    2. 查询数据
      result = sparkSession.sql("SELECT * FROM opentsdb_test")
  • 通过DataFrame API 访问
    1. 构造schema
      1
      2
      3
      4
      schema = StructType([StructField("location", StringType()),                     
                           StructField("name", StringType()),                    
                           StructField("timestamp", LongType()),                  
                           StructField("value", DoubleType())])
      
    2. 设置数据
      1
      dataList = sparkSession.sparkContext.parallelize([("beijing", "abc", 123456L, 30.0)])
      
    3. 创建DataFrame
      1
      dataFrame = sparkSession.createDataFrame(dataList, schema)
      
    4. 导入数据到OpenTSDB
      1
      dataFrame.write.insertInto("opentsdb_test")
      
    5. 读取OpenTSDB上的数据
      1
      2
      3
      4
      5
      6
      7
      jdbdDF = sparkSession.read
          .format("opentsdb")\
          .option("Host","opentsdb-3xcl8dir15m58z3.cloudtable.com:4242")\
          .option("metric","ctopentsdb")\
          .option("tags","city,location")\
          .load()
      jdbdDF.show()
      
    6. 操作结果

  • 提交Spark作业
    1. 将写好的python代码文件上传至DLI中。控制台操作请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>《上传资源包》。
    2. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。控制台操作请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>《创建批处理作业》。
      • 提交作业时,需要指定Module模块,名称为:sys.datasource.opentsdb。
      • 通过控制台提交作业请参考《数据湖探索用户指南》中的“选择依赖资源参数说明”表说明。
      • 通过API提交作业请参考《数据湖探索API参考》>《创建批处理作业》中“表2-请求参数说明”关于“modules”参数的说明。

完整示例代码

  • 通过SQL API访问MRS的OpenTSDB
     1
     2
     3
     4
     5
     6
     7
     8
     9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    # _*_ 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('beijing', '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([("beijing", "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()
分享:

    相关文档

    相关产品

close