Spark Streaming之Transform算子详解

Myth丶恋晨 2022-12-28 12:52 156阅读 0赞

**Transform算子解读：**  
transform操作，应用在DStream上时，可以用于执行任意的RDD到RDD的转换操作。它可以用于实现， DStream API中所没有提供的操作。比如说，DStream API中，并没有提供将一个DStream中的每个 batch，与一个特定的RDD进行join的操作。但是我们自己就可以使用transform操作来实现该功能。

DStream.join()，只能join其他DStream。在DStream每个batch的RDD计算出来之后，会去跟其他 DStream的RDD进行join。

**案例：广告计费日志实时黑名单过滤**

需求分析：  
实时过滤出黑名单的人员

数据模型：

1.黑名单数据 
    Array((“张三”,true))
    
    2.stream数据 
    2019-09-09 张三 
    2019-09-09 李四 
    2019-09-09 王五
    
    3.stream数据格式转换
    
    <“张三”, 2019-09-09 张三>
    
    4.Transform
    <"张三",2019-09-09 张三> <"张三",true> -> leftOuterJoin -> 
    <”张三”,<张三 2019-09-09,true>> -> filter ->
    过滤掉==true的数据

实施过程：  
Java语言实现：

package com.kfk.spark.common;
    
    import org.apache.spark.SparkConf;
    import org.apache.spark.streaming.Durations;
    import org.apache.spark.streaming.api.java.JavaStreamingContext;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/12/14 * @time : 8:23 下午 */
    public class CommStreamingContext { 
    
        public static JavaStreamingContext getJssc(){ 
            SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("CommStreamingContext");
            return new JavaStreamingContext(conf, Durations.seconds(2));
        }
    }

package com.kfk.spark.blacklist_project;
    
    import com.kfk.spark.common.CommStreamingContext;
    import org.apache.spark.api.java.JavaPairRDD;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.Optional;
    import org.apache.spark.api.java.function.Function;
    import org.apache.spark.api.java.function.PairFunction;
    import org.apache.spark.streaming.api.java.JavaDStream;
    import org.apache.spark.streaming.api.java.JavaPairDStream;
    import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
    import org.apache.spark.streaming.api.java.JavaStreamingContext;
    import scala.Tuple2;
    
    import java.util.ArrayList;
    import java.util.List;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/12/16 * @time : 8:21 下午 */
    public class BlackListJava { 
        public static void main(String[] args) throws InterruptedException { 
            JavaStreamingContext jssc = CommStreamingContext.getJssc();
    
            /** * 模拟创建黑名单 * 数据格式：<String,Boolean> -> <"jack",true> */
            List<Tuple2<String,Boolean>> blacklist = new ArrayList<Tuple2<String,Boolean>>();
            blacklist.add(new Tuple2<>("jack",true));
            blacklist.add(new Tuple2<>("lili",true));
    
            JavaPairRDD<String,Boolean> blacklistRdd = jssc.sparkContext().parallelizePairs(blacklist);
    
            // nc数据源
            JavaReceiverInputDStream<String> userDstream = jssc.socketTextStream("bigdata-pro-m04",9999);
    
            /** * 源数据：2019-09-09 张三 -> 转换后的数据： <"张三",2019-09-09 张三> */
            JavaPairDStream<String, String> userPair = userDstream.mapToPair(new PairFunction<String, String, String>() { 
                @Override
                public Tuple2<String, String> call(String line) throws Exception { 
                    return new Tuple2<String, String>(line.split(" ")[1],line);
                }
            });
    
            /** * <"张三",2019-09-09 张三> <"张三",true> -> leftOuterJoin -> <”张三”,<张三 2019-09-09,true>> -> filter 过滤掉==true的数据 */
            JavaDStream<String> stream = userPair.transform(new Function<JavaPairRDD<String, String>, JavaRDD<String>>() { 
                @Override
                public JavaRDD<String> call(JavaPairRDD<String, String> pair) throws Exception { 
    
                    // leftOuterJoin
                    JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> joinrdd = pair.leftOuterJoin(blacklistRdd);
    
                    // filter
                    JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> filterjoinrdd =joinrdd.filter(
                            new Function<Tuple2<String, Tuple2<String, Optional<Boolean>>>, Boolean>() { 
                        @Override
                        public Boolean call(Tuple2<String, Tuple2<String, Optional<Boolean>>> tuple) throws Exception { 
    
                            if (tuple._2._2.isPresent() && tuple._2._2.get()){ 
                                return false;
                            }
                            return true;
                        }
                    });
    
                    // map
                    JavaRDD<String> rdd = filterjoinrdd.map(new Function<Tuple2<String,
                            Tuple2<String, Optional<Boolean>>>, String>() { 
                        @Override
                        public String call(Tuple2<String, Tuple2<String, Optional<Boolean>>> tuple) throws Exception { 
                            return tuple._1 + " : " + tuple._2._1 + " : " + tuple._2._2;
                        }
                    });
    
                    return rdd;
                }
            });
    
            stream.print();
    
            jssc.start();
            jssc.awaitTermination();
    
        }
    }

Scala语言实现：

package com.kfk.spark.blacklist_project
    
    import com.kfk.spark.common.CommStreamingContextScala
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/12/16 * @time : 10:31 下午 */
    object BlackListScala { 
        def main(args: Array[String]): Unit = { 
            val jssc = CommStreamingContextScala.getJssc
    
            /** * 模拟创建黑名单 * 数据格式：<String,Boolean> -> <"jack",true> */
            val blacklist = Array(("ben",true),("leo",true))
    
            val blacklistRdd = jssc.sparkContext.parallelize(blacklist)
    
            // nc数据源
            val userDstream = jssc.socketTextStream("bigdata-pro-m04", 9999)
    
            /** * 源数据：2019-09-09 张三 -> 转换后的数据： <"张三",2019-09-09 张三> */
            val pair = userDstream.map(line => (line.split(" ")(1),line))
    
            /** * <"张三",2019-09-09 张三> <"张三",true> -> leftOuterJoin -> <”张三”,<张三 2019-09-09,true>> -> filter 过滤掉==true的数据 */
            val vailStream = pair.transform(tuple => { 
                val leftJoinRDD = tuple.leftOuterJoin(blacklistRdd)
                val filterRDD = leftJoinRDD.filter(rdd => { 
                    if (rdd._2._2.getOrElse(0) == true) false else true
                })
    
                val valiRDD = filterRDD.map(tuple => { 
                    tuple._1 + " : " + tuple._2._1 + " : " + tuple._2._2
                })
                valiRDD
            })
    
            vailStream.print()
    
            jssc.start()
            jssc.awaitTermination()
        }
    }

测试数据

2019-09-09 jack
    2019-09-09 jone
    2019-09-09 lili
    2019-09-09 alex
    2019-09-09 cherry
    2019-09-09 carry
    2019-09-09 pony
    2019-09-09 jack

黑名单数据：

2019-09-09 jack
    2019-09-09 lili

运行结果：

-------------------------------------------
    Time: 1608178954000 ms
    -------------------------------------------
    alex : 2019-09-09 alex : Optional.empty
    pony : 2019-09-09 pony : Optional.empty
    jone : 2019-09-09 jone : Optional.empty
    carry : 2019-09-09 carry : Optional.empty
    cherry : 2019-09-09 cherry : Optional.empty

可以看出黑名单数据已经被过滤掉了。

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！