数据湖探索 DLI-pyspark样例代码:完整示例代码

时间：2024-11-16 13:16:39

数据湖探索 DLI

完整示例代码

通过DataFrame API访问

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

    
     
       
       
         # _*_ 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访问

    
         # _*_ 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()