# pyspark样例代码
#### 开发说明
redis只支持增强型跨源。只能使用包年包月队列。
- 前提条件 在DLI管理控制台上已完成创建增强跨源连接，并绑定包年包月队列。具体操作请参考《[数据湖探索用户指南](https://support.huaweicloud.com/usermanual-dli/dli_01_0426.html)》。
  ![](https://support.huaweicloud.com/devg-dli/public_sys-resources/note_3.0-zh-cn.png)
  认证用的password硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。
  
- 通过DataFrame API 访问
  1. import相关依赖
     ```
     from __future__ import print_function
     from pyspark.sql.types import StructType, StructField, IntegerType, StringType
     from pyspark.sql import SparkSession
     ```
     
  
  2. 创建session
     ```
     sparkSession = SparkSession.builder.appName("datasource-redis").getOrCreate()
     ```
     
  
  3. 设置连接参数
     ```
     host = "192.168.4.199"
     port = "6379"
     table = "person"
     auth = "@@@@@@"
     ```
     
  
  4. 创建DataFrame
     1. 方式一
        ```
        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. 方式二
        ```
        jdbcDF = sparkSession.createDataFrame([(3,"Jack", 23)])
        dataFrame = jdbcDF.withColumnRenamed("_1", "id").withColumnRenamed("_2", "name").withColumnRenamed("_3", "age")
        ```
        
      
  
  5. 导入数据到redis
     ```
     dataFrame.write
       .format("redis")\
       .option("host", host)\
       .option("port", port)\
       .option("table", table)\
       .option("password", auth)\
       .mode("Overwrite")\
       .save()
     ```
     ![](https://support.huaweicloud.com/devg-dli/public_sys-resources/note_3.0-zh-cn.png)
     - 保存类型：Overwrite、Append、ErrorIfExists、Ignore 四种
     
     - 如果需要指定key，则通过".option("key.column","name")"指定，name为列名
     
     - 如果需要保存嵌套的DataFrame，则通过".option("model","binary")"进行保存
     
     - 如果需要指定数据过期时间：".option("ttl",1000)";秒为单位
       
  
  6. 读取redis上的数据
     ```
     sparkSession.read.format("redis").option("host", host).option("port", port).option("table", table).option("password", auth).load().show()
     ```
     
  
  7. 操作结果 ![](https://support.huaweicloud.com/devg-dli/zh-cn_image_0223997787.png)
     
   
- 通过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. 插入数据
     ```
     sparkSession.sql("INSERT INTO TABLE person VALUES ('John', 30),('Peter', 45)".stripMargin)
     ```
     
  
  3. 查询数据
     ```
     sparkSession.sql("SELECT * FROM person".stripMargin).collect().foreach(println)
     ```
     
   
- 提交Spark作业
  1. 将写好的python代码文件上传至OBS桶中。
  
  2. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。
     ![](https://support.huaweicloud.com/devg-dli/public_sys-resources/note_3.0-zh-cn.png)
     - 如果选择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/\*
       
     
     - 通过控制台提交作业请参考《[数据湖探索用户指南](https://support.huaweicloud.com/usermanual-dli/dli_01_0384.html)》中的"选择依赖资源参数说明"。
     
     - 通过API提交作业请参考《数据湖探索API参考》\>《[创建批处理作业](https://support.huaweicloud.com/api-dli/dli_02_0124.html)》中"表2-请求参数说明"关于"modules"参数的说明。
       
   
 
#### 完整示例代码
- 通过DataFrame API 访问
  ```
  # _*_ 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 访问
  ```
  # _*_ 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()
  ```