def writeRedis(sc: SparkContext,spark: SparkSession): Unit ={
    // spark讀取數(shù)據(jù)集
     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"))
       // 這個(gè)集合寫(xiě)的是2000多萬(wàn)的數(shù)據(jù)
    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()
        // 選擇數(shù)據(jù)庫(kù)  默認(rèn)為 0
        pipeline.select(1)
        for(i <- 0 to (num - 1) ){
          pipeline.srem(collectionName,arr(i))
        }
        //表示我一次性的異步發(fā)送到redis，不關(guān)注執(zhí)行結(jié)果
        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()
      // 驅(qū)動(dòng)進(jìn)程使用的內(nèi)核數(shù)，僅在集群模式下使用。
      .set("spark.driver.cores","5")
      /**
        * 驅(qū)動(dòng)進(jìn)程使用的內(nèi)存數(shù)量，也就是SparkContext初始化的地方，
        * 其格式與JVM內(nèi)存字符串具有大小單位后綴(“k”，“m”，“g”或“t”)(例如512m, 2g)相同。
        * 注意:在客戶端模式下，不能直接在應(yīng)用程序中通過(guò)SparkConf設(shè)置此配置，因?yàn)榇藭r(shí)驅(qū)動(dòng)程
        * 序JVM已經(jīng)啟動(dòng)。相反，請(qǐng)通過(guò)——driver-memory命令行選項(xiàng)或在默認(rèn)屬性文件中設(shè)置。
        */
      .set("spark.driver.memory","5g")
      /**
        * 限制每個(gè)Spark操作(例如collect)的所有分區(qū)的序列化結(jié)果的總大小(以字節(jié)為單位)。
        * 應(yīng)該至少是1M，或者0表示無(wú)限制。如果總大小超過(guò)此限制，則作業(yè)將被終止。
        * 過(guò)高的限制可能會(huì)導(dǎo)致驅(qū)動(dòng)程序內(nèi)存不足錯(cuò)誤(取決于spark.driver.memory和JVM中對(duì)象的內(nèi)存開(kāi)銷)。
        * 設(shè)置適當(dāng)?shù)南拗瓶梢苑乐跪?qū)動(dòng)程序出現(xiàn)內(nèi)存不足的錯(cuò)誤。
        */
      .set("spark.driver.maxResultSize","10g")
      /**
        * 每個(gè)執(zhí)行程序進(jìn)程使用的內(nèi)存數(shù)量，
        * 格式與帶有大小單位后綴(“k”，“m”，“g”或“t”)的JVM內(nèi)存字符串相同(例如512m, 2g)。
        *
        */
      .set("spark.executor.memory","5g")
      /**
        * 默認(rèn) 1在YARN模式下，worker上所有可用的內(nèi)核在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")     // 可選的數(shù)據(jù)庫(kù)編號(hào)。避免使用它，尤其是在集群模式下,redisRedis默認(rèn)支持16個(gè)數(shù)據(jù)庫(kù)，默認(rèn)是選擇數(shù)據(jù)庫(kù)0，這里設(shè)置為1。
      .config("spark.redis.timeout","2000000")   // 連接超時(shí)，以毫秒為單位，默認(rèn)為 2000 毫秒
      .config(conf)
      .getOrCreate()
    val sc: SparkContext = spark.sparkContext
     //1.寫(xiě)入數(shù)據(jù)集
   writeRedis(sc,spark)
    // 2.讀取待刪除的數(shù)據(jù)key
      val path = "file:///F://delRedisData//test.parquet"
    val rdd: RDD[String] = readParquet(spark,path)
    //3.使用redis 中的 pipeline 方法 進(jìn)行刪除操作
    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+"條數(shù)據(jù)，耗時(shí)："+(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()
        // 選擇數(shù)據(jù)庫(kù)  默認(rèn)為 0
        pipeline.select(1)
        for(i <- 0 to (num - 1) ){
          pipeline.srem(collectionName,arr(i))
        }
        //表示我一次性的異步發(fā)送到redis，不關(guān)注執(zhí)行結(jié)果
        pipeline.sync()
    }catch {
      case e : JedisException => e.printStackTrace()
    }finally if(jedis !=null) jedis.close()
  }
def writeRedis(sc: SparkContext,spark: SparkSession): Unit ={
    // spark讀取數(shù)據(jù)集
    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"))
       // 這個(gè)集合寫(xiě)的是2000多萬(wàn)的數(shù)據(jù)
    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
  }
  }