文档首页> 数据湖探索 DLI> 开发指南> Spark作业开发指南> 使用Spark作业跨源访问数据源> 对接DWS> pyspark样例代码
更新时间：2022-04-24 GMT+08:00
查看PDF
分享

pyspark样例代码

开发说明

  • 前提条件

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

  • 代码实现详解
    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参考》>《创建批处理作业》。
      • 提交作业时，需要指定Module模块，名称为：sys.datasource.dws。
      • 通过控制台提交作业请参考《数据湖探索用户指南》中的“选择依赖资源参数说明”表说明。
      • 通过API提交作业请参考《数据湖探索API参考》>《创建批处理作业》中“表2-请求参数说明”关于“modules”参数的说明。

完整示例代码

  • 通过DataFrame 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
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()
父主题： 对接DWS
分享：

    相关文档

    相关产品

close