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

pyspark样例代码

开发说明

redis只支持增强型跨源。

  • 前提条件

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

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

  • 通过DataFrame API 访问
    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. 创建session
      1
      sparkSession = SparkSession.builder.appName("datasource-redis").getOrCreate()
      
    3. 设置连接参数
      1
      2
      3
      4
      host = "192.168.4.199"
      port = "6379"
      table = "person"
      auth = "@@@@@@"
      
    4. 创建DataFrame
      1. 方式一
        1
        2
        3
        4
        5
        dataList = sparkSession.sparkContext.parallelize([(1, "Katie", 19),(2,"Tom",20)])
        schema = StructType([StructField("id", IntegerType(), False),          
                             StructField("name", StringType(), False),
                             StructField("age", IntegerType(), False)])
        dataFrame = sparkSession.createDataFrame(dataList, schema)
        
      2. 方式二
        1
        2
        jdbcDF = sparkSession.createDataFrame([(3,"Jack", 23)])
        dataFrame = jdbcDF.withColumnRenamed("_1", "id").withColumnRenamed("_2", "name").withColumnRenamed("_3", "age")
        
    5. 导入数据到redis
      1
      2
      3
      4
      5
      6
      7
      8
      dataFrame.write
        .format("redis")\
        .option("host", host)\
        .option("port", port)\
        .option("table", table)\
        .option("password", auth)\
        .mode("Overwrite")\
        .save()
      
      • 保存类型:Overwrite、Append、ErrorIfExis、Ignore 四种
      • 如果需要指定key,则通过“.option("key.column","name")”指定,name为列名
      • 如果需要保存嵌套的DataFrame,则通过“.option("model","binary")”进行保存
      • 如果需要指定数据过期时间:“.option("ttl",1000)”;秒为单位
    6. 读取redis上的数据
      1
      sparkSession.read.format("redis").option("host", host).option("port", port).option("table", table).option("password", auth).load().show()
      
    7. 操作结果

  • 通过SQL API 访问
    1. 创建DLI关联跨源访问 Redis的关联表。
      sparkSession.sql(
           "CREATE TEMPORARY VIEW person (name STRING, age INT) USING org.apache.spark.sql.redis OPTIONS (
           'host' = '192.168.4.199',
           'port' = '6379',
           'password' = '######',
           table  'person')".stripMargin)
    2. 插入数据
      1
      sparkSession.sql("INSERT INTO TABLE person VALUES ('John', 30),('Peter', 45)".stripMargin)
      
    3. 查询数据
      1
      sparkSession.sql("SELECT * FROM person".stripMargin).collect().foreach(println)
      
  • 提交Spark作业
    1. 将写好的python代码文件上传至DLI中。

      控制台操作请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>《上传资源包》。

    2. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。

      控制台操作请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>《创建批处理作业》。
      • 如果选择spark版本为2.3.2(即将下线)或2.4.5提交作业时,需要指定Module模块,名称为:sys.datasource.redis。
      • 如果选择Spark版本为3.1.1时,无需选择Module模块, 需在 'Spark参数(--conf)' 配置

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

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

      • 通过控制台提交作业请参考《数据湖探索用户指南》中的“选择依赖资源参数说明”表说明
      • 通过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
    # _*_ 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-redis").getOrCreate()
      
      # Set cross-source connection parameters.
      host = "192.168.4.199"
      port = "6379"
      table = "person"  
      auth = "######"
         
      # Create a DataFrame and initialize the DataFrame data.    
      # *******   method noe   *********    
      dataList = sparkSession.sparkContext.parallelize([(1, "Katie", 19),(2,"Tom",20)])
      schema = StructType([StructField("id", IntegerType(), False),StructField("name", StringType(), False),StructField("age", IntegerType(), False)])
      dataFrame_one = sparkSession.createDataFrame(dataList, schema)
    
      # ****** method two ******  
      # jdbcDF = sparkSession.createDataFrame([(3,"Jack", 23)])
      # dataFrame = jdbcDF.withColumnRenamed("_1", "id").withColumnRenamed("_2", "name").withColumnRenamed("_3", "age")
      
      # Write data to the redis table  
      dataFrame.write.format("redis").option("host", host).option("port", port).option("table", table).option("password", auth).mode("Overwrite").save()  
      # Read data  
      sparkSession.read.format("redis").option("host", host).option("port", port).option("table", table).option("password", auth).load().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
    # _*_ coding: utf-8 _*_
    from __future__ import print_function
    from pyspark.sql import SparkSession
    
    if __name__ == "__main__":
      # Create a SparkSession  
      sparkSession = SparkSession.builder.appName("datasource_redis").getOrCreate()
    
      sparkSession.sql(
        "CREATE TEMPORARY VIEW person (name STRING, age INT) USING org.apache.spark.sql.redis OPTIONS (\
        'host' = '192.168.4.199', \
        'port' = '6379',\
        'password' = '######',\
        'table'= 'person')".stripMargin); 
    
      sparkSession.sql("INSERT INTO TABLE person VALUES ('John', 30),('Peter', 45)".stripMargin)
      
      sparkSession.sql("SELECT * FROM person".stripMargin).collect().foreach(println)
       
      # close session  
      sparkSession.stop()