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

pyspark样例代码

开发说明

redis只支持增强型跨源。只能使用包年包月队列。

  • 前提条件

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

  • 通过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的关联表。
      1
2
3
4
5
6
      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参考》>《创建批处理作业》。
      • 提交作业时，需要指定Module模块，名称为：sys.datasource.redis。
      • 通过控制台提交作业请参考《数据湖探索用户指南》中的“表6-选择依赖资源参数说明”。
      • 通过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
    # _*_ 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
分享：

    相关文档

    相关产品

close