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

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()
父主题： 对接Redis