Spark删除redis千万级别set集合数据实现分析

目录

• 1.使用pipline的原因
• 2.方法
• 2.1写入redis的方法
• 2.1.1参数说明
• 2.2读取本地待删除数据的方法
• 2.2.1参数说明
• 2.3调用pipline删除的方法
• 2.3.1参数说明
• 3.完整代码
• 4.总结

def writeRedis(sc: SparkContext,spark: SparkSession): Unit ={
    // spark读取数据集
     val df: DataFrame = spark.read.parquet("file:///F://delRedisData//1//delData.snappy.parquet")
    df.show(1,false)
    val rdd: RDD[String] = df.rdd.map(x=>x.getAs[String]("r"))
       // 这个集合写的是2000多万的数据
    sc.toRedisSET(rdd,"test:task:deplicate")
  }

def readParquet(spark: SparkSession,path:String): RDD[String] ={
    val df: DataFrame = spark.read.parquet(path)
    val strRDD: RDD[String] = df.rdd.map(_.getAs[String]("r"))
    // 返回String类型的RDD
    strRDD
  }

def delPipleine(collectionName:String,num:Int,arr:Array[String],jedis:Jedis):Unit = {
    try{
        val pipeline: Pipeline = jedis.pipelined()
        // 选择数据库  默认为 0
        pipeline.select(1)
        for(i <- 0 to (num - 1) ){
          pipeline.srem(collectionName,arr(i))
        }
        //表示我一次性的异步发送到redis，不关注执行结果
        pipeline.sync()
    }catch {
      case e : JedisException => e.printStackTrace()
    }finally if(jedis !=null) jedis.close()
  }

import com.redislabs.provider.redis._
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
import redis.clients.jedis.exceptions.JedisException
import redis.clients.jedis.{Jedis, Pipeline}
/**
  * Date 2022/5/25 17:57
  */
object DelRedis {
def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      // 驱动进程使用的内核数，仅在集群模式下使用。
      .set("spark.driver.cores","5")
      /**
        * 驱动进程使用的内存数量，也就是SparkContext初始化的地方，
        * 其格式与JVM内存字符串具有大小单位后缀(“k”，“m”，“g”或“t”)(例如512m, 2g)相同。
        * 注意:在客户端模式下，不能直接在应用程序中通过SparkConf设置此配置，因为此时驱动程
        * 序JVM已经启动。相反，请通过——driver-memory命令行选项或在默认属性文件中设置。
        */
      .set("spark.driver.memory","5g")
      /**
        * 限制每个Spark操作(例如collect)的所有分区的序列化结果的总大小(以字节为单位)。
        * 应该至少是1M，或者0表示无限制。如果总大小超过此限制，则作业将被终止。
        * 过高的限制可能会导致驱动程序内存不足错误(取决于spark.driver.memory和JVM中对象的内存开销)。
        * 设置适当的限制可以防止驱动程序出现内存不足的错误。
        */
      .set("spark.driver.maxResultSize","10g")
      /**
        * 每个执行程序进程使用的内存数量，
        * 格式与带有大小单位后缀(“k”，“m”，“g”或“t”)的JVM内存字符串相同(例如512m, 2g)。
        *
        */
      .set("spark.executor.memory","5g")
      /**
        * 默认 1在YARN模式下，worker上所有可用的内核在standalone和Mesos粗粒度模式下。
        */
      .set("spark.executor.cores","5")
    val spark: SparkSession = SparkSession.builder().appName("DelRedis").master("local[*]")
      .config("spark.redis.host","192.168.100.201")
      .config("spark.redis.port","6379")
      .config("spark.redis.db","1")     // 可选的数据库编号。避免使用它，尤其是在集群模式下,redisRedis默认支持16个数据库，默认是选择数据库0，这里设置为1。
      .config("spark.redis.timeout","2000000")   // 连接超时，以毫秒为单位，默认为 2000 毫秒
      .config(conf)
      .getOrCreate()
    val sc: SparkContext = spark.sparkContext
     //1.写入数据集
   writeRedis(sc,spark)
    // 2.读取待删除的数据key
      val path = "file:///F://delRedisData//test.parquet"
    val rdd: RDD[String] = readParquet(spark,path)
    //3.使用redis 中的 pipeline 方法 进行删除操作
    rdd.foreachPartition(iter=>{
      // 连接redis客户端
      val jedis = new Jedis("192.168.100.201",6379)
      val array: Array[String] = iter.toArray
      val length: Int = array.length
      val beginTime: Long = System.currentTimeMillis()
      delPipleine(collectionName,length,array,jedis)
      val endTime: Long = System.currentTimeMillis()
      println("删除："+length+"条数据，耗时："+(endTime-beginTime)/1000+"秒")
    })
    sc.stop()
    spark.stop()
  }
def delPipleine(collectionName:String,num:Int,arr:Array[String],jedis:Jedis):Unit = {
    try{
        val pipeline: Pipeline = jedis.pipelined()
        // 选择数据库  默认为 0
        pipeline.select(1)
        for(i <- 0 to (num - 1) ){
          pipeline.srem(collectionName,arr(i))
        }
        //表示我一次性的异步发送到redis，不关注执行结果
        pipeline.sync()
    }catch {
      case e : JedisException => e.printStackTrace()
    }finally if(jedis !=null) jedis.close()
  }
def writeRedis(sc: SparkContext,spark: SparkSession): Unit ={
    // spark读取数据集
    val df: DataFrame = spark.read.parquet("file:///F://delRedisData//1//delData.snappy.parquet")
    df.show(1,false)
    val rdd: RDD[String] = df.rdd.map(x=>x.getAs[String]("r"))
       // 这个集合写的是2000多万的数据
    sc.toRedisSET(rdd,"test:task:deplicate")
  }
def readParquet(spark: SparkSession,path:String): RDD[String] ={
    val df: DataFrame = spark.read.parquet(path)
    val strRDD: RDD[String] = df.rdd.map(_.getAs[String]("r"))
    // 返回String类型的RDD
    strRDD
  }
  }