更新时间:2024-07-04 GMT+08:00

pyspark样例代码

操作场景

本例提供使用Spark作业访问DWS数据源的pyspark样例代码。

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

认证用的password硬编码到代码中或者明文存储都有很大的安全风险,建议在配置文件或者环境变量中密文存放,使用时解密,确保安全。

操作前准备

  1. import相关依赖包
    1
    2
    3
    from __future__ import print_function
    from pyspark.sql.types import StructType, StructField, IntegerType, StringType
    from pyspark.sql import SparkSession
    
  2. 创建会话
    1
    sparkSession = SparkSession.builder.appName("datasource-dws").getOrCreate()
    

通过DataFrame API访问数据源

  1. 连接参数配置
    1
    2
    3
    4
    5
    url = "jdbc:postgresql://to-dws-1174404951-W8W4cW8I.datasource.com:8000/postgres"
    dbtable = "customer"
    user = "dbadmin"
    password = "######"
    driver = "org.postgresql.Driver"
    
  2. 设置数据
    1
    dataList = sparkSession.sparkContext.parallelize([(1, "Katie", 19)])
    
  3. 设置schema
    1
    2
    3
    schema = StructType([StructField("id", IntegerType(), False),\                
                         StructField("name", StringType(), False),\            
                         StructField("age", IntegerType(), False)])
    
  4. 创建DataFrame
    1
    dataFrame = sparkSession.createDataFrame(dataList, schema)
    
  5. 保存数据到DWS
    1
    2
    3
    4
    5
    6
    7
    8
    9
    dataFrame.write \   
        .format("jdbc") \  
        .option("url", url) \  
        .option("dbtable", dbtable) \  
        .option("user", user) \ 
        .option("password", password) \ 
        .option("driver", driver) \ 
        .mode("Overwrite") \  
        .save()
    

    mode 有四种保存类型:

    • ErrorIfExis:如果已经存在数据,则抛出异常。
    • Overwrite:如果已经存在数据,则覆盖原数据。
    • Append:如果已经存在数据,则追加保存。
    • Ignore:如果已经存在数据,则不做操作。这类似于SQL中的“如果不存在则创建表”。
  6. 读取DWS上的数据
    1
    2
    3
    4
    5
    6
    7
    8
    9
    jdbcDF = sparkSession.read \
        .format("jdbc") \
        .option("url", url) \
        .option("dbtable", dbtable) \
        .option("user", user) \
        .option("password", password) \
        .option("driver", driver) \
        .load()
    jdbcDF.show()
    
  7. 操作结果

通过SQL API 访问数据源

  1. 创建DLI跨源访问 dws 的关联表。
    1
    2
    3
    4
    5
    6
    7
    sparkSession.sql(
        "CREATE TABLE IF NOT EXISTS dli_to_dws USING JDBC OPTIONS (
        'url'='jdbc:postgresql://to-dws-1174404951-W8W4cW8I.datasource.com:8000/postgres',\
        'dbtable'='customer',\
        'user'='dbadmin',\
        'password'='######',\
        'driver'='org.postgresql.Driver')")
    

    建表参数详情可参考表1

  2. 插入数据
    1
    sparkSession.sql("insert into dli_to_dws values(2,'John',24)")
    
  3. 查询数据
    1
    jdbcDF = sparkSession.sql("select * from dli_to_dws").show()
    
  4. 操作结果

提交Spark作业

  1. 将写好的python代码文件上传至DLI中。控制台操作请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>《上传资源包》。
  2. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。控制台操作请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>《创建批处理作业》。
    • 如果选择spark版本为2.3.2(即将下线)或2.4.5提交作业时,需要指定Module模块,名称为:sys.datasource.hbase。
    • 如果选择Spark版本为3.1.1时,无需选择Module模块, 需在 'Spark参数(--conf)' 配置

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

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

    • 通过控制台提交作业请参考《数据湖探索用户指南》中的“选择依赖资源参数说明”表说明
    • 通过API提交作业请参考《数据湖探索API参考》>《创建批处理作业》中“表2-请求参数说明”关于“modules”参数的说明。

完整示例代码

  • 通过DataFrame API访问

    认证用的password硬编码到代码中或者明文存储都有很大的安全风险,建议在配置文件或者环境变量中密文存放,使用时解密,确保安全。

     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
    # _*_ coding: utf-8 _*_
    from __future__ import print_function
    from pyspark.sql.types import StructType, StructField, IntegerType, StringType
    from pyspark.sql import SparkSession
    
    if __name__ == "__main__":
      # Create a SparkSession session.   
      sparkSession = SparkSession.builder.appName("datasource-dws").getOrCreate()
    
      # Set cross-source connection parameters  
      url = "jdbc:postgresql://to-dws-1174404951-W8W4cW8I.datasource.com:8000/postgres"
      dbtable = "customer" 
      user = "dbadmin"
      password = "######"
      driver = "org.postgresql.Driver"
    
      # Create a DataFrame and initialize the DataFrame data.   
      dataList = sparkSession.sparkContext.parallelize([(1, "Katie", 19)])
    
      # Setting schema   
      schema = StructType([StructField("id", IntegerType(), False),\     
                           StructField("name", StringType(), False),\    
                           StructField("age", IntegerType(), False)])
    
      # Create a DataFrame from RDD and schema   
      dataFrame = sparkSession.createDataFrame(dataList, schema)
    
      # Write data to the DWS table  
      dataFrame.write \ 
          .format("jdbc") \    
          .option("url", url) \  
          .option("dbtable", dbtable) \    
          .option("user", user) \    
          .option("password", password) \   
          .option("driver", driver) \     
          .mode("Overwrite") \   
          .save()
    
      # Read data   
      jdbcDF = sparkSession.read \  
          .format("jdbc") \     
          .option("url", url) \  
          .option("dbtable", dbtable) \     
          .option("user", user) \     
          .option("password", password) \   
          .option("driver", driver) \  
          .load()  
      jdbcDF.show()
    
      # close session  
      sparkSession.stop()
    
  • 通过SQL API访问
     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
    # _*_ 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-dws").getOrCreate()
    
      # Createa data table for DLI - associated DWS 
      sparkSession.sql(
          "CREATE TABLE IF NOT EXISTS dli_to_dws USING JDBC OPTIONS (\
          'url'='jdbc:postgresql://to-dws-1174404951-W8W4cW8I.datasource.com:8000/postgres',\
          'dbtable'='customer',\
          'user'='dbadmin',\
          'password'='######',\
          'driver'='org.postgresql.Driver')")
    
      # Insert data into the DLI data table  
      sparkSession.sql("insert into dli_to_dws values(2,'John',24)")
    
      # Read data from DLI data table  
      jdbcDF = sparkSession.sql("select * from dli_to_dws").show()
    
      # close session  
      sparkSession.stop()