文章地址：http://www.haha174.top/article/details/254946 transform 操作，应用在DStream 上时，可以用于执行任意的RDD 到RDD 转换的操作。他可以用于实现，DStream api 所没有提供的操作。比如说，DStream api 中并没有提供一个DStream 中的每个batch ，与一个特定的RDD 进行join 操作。但是我们自己就可以使用transform 操作来实现该功能。 DStream.join（） 只能join 其他的DStream ,在DStream 每个 batch 的RDD 计算出来之后，会去跟其他的DStream 的RDD 行join 下面给出java 示例和注释：
public class TransFormBlackList { public static void main(String[] args) throws InterruptedException { System.setProperty("HADOOP_USER_NAME", "root"); SparkConf conf=new SparkConf().setMaster("local[2]").setAppName("TransFormBlackList"); JavaSparkContext sc=new JavaSparkContext(conf); JavaStreamingContext jssc=new JavaStreamingContext(sc, Durations.seconds(5)); // 用户对我们的网站上面的广告可以进行点击 //点击之后，要进行实时的计费，点一下，算一次钱 //但是对于某些无良商家刷广告的人，那么我们就有一个黑名单 //只要是黑名单中点击的都过滤掉 // 日志格式 date username // 先做一份模拟的黑名单 List<Tuple2<String,Boolean>> blackListData=new ArrayList<>(); blackListData.add(new Tuple2<>("tom",true)); final JavaPairRDD<String,Boolean> blackListRDD=sc.parallelizePairs(blackListData); JavaReceiverInputDStream<String> adsClickLog=jssc.socketTextStream("www.codeguoj.cn",9999); // 首先需要转换一个格式(date,date username) //以便于后面 于黑名单RDD 进行join JavaPairDStream<String,String> userAdsClickDStream=adsClickLog.mapToPair(new PairFunction<String, String, String>() { @Override public Tuple2<String, String> call(String s) throws Exception { return new Tuple2<>(s.split(" ")[1],s); } }); //执行transform 操作实时进行黑名单过滤 JavaDStream<String> validAdfClickLogDStream= userAdsClickDStream.transform(new Function<JavaPairRDD<String, String>, JavaRDD<String>>() { private static final long serialVersionUID=1L; @Override public JavaRDD<String> call(JavaPairRDD<String, String> v1) throws Exception { // 并不是每个用户都存在给名单中所以使用做外链接 JavaPairRDD<String, Tuple2<String, org.apache.spark.api.java.Optional<Boolean>>> joindRDD= v1.leftOuterJoin(blackListRDD); //这里得到的是用户的点击日志 和在黑名单的状态 JavaPairRDD<String, Tuple2<String, org.apache.spark.api.java.Optional<Boolean>>> filterRDD=joindRDD.filter(new Function<Tuple2<String, Tuple2<String, org.apache.spark.api.java.Optional<Boolean>>>, Boolean>() { private static final long serialVersionUID=1L; @Override public Boolean call(Tuple2<String, Tuple2<String, org.apache.spark.api.java.Optional<Boolean>>> v1) throws Exception { if(v1._2._2().isPresent()&&v1._2._2().get()){ return false; } return true; } }); JavaRDD<String> validAdfClickRDD=filterRDD.map(new Function<Tuple2<String, Tuple2<String, org.apache.spark.api.java.Optional<Boolean>>>, String>() { @Override public String call(Tuple2<String, Tuple2<String, org.apache.spark.api.java.Optional<Boolean>>> v1) throws Exception { return v1._2._1; } }); return validAdfClickRDD; } }); validAdfClickLogDStream.print(); jssc.start(); jssc.awaitTermination(); jssc.stop(); } }下面给出scala 示例和注释
object TransFormBlackList { def main(args: Array[String]): Unit = { val conf = new SparkConf().setAppName("updateStateByKeyWorldCount").setMaster("local[2]") val jssc = new StreamingContext(conf, Durations.seconds(5)) //首先创建输入DStream 代表一个数据源 （比如kafka ,socket） 来持续不断的实时数据流 //调用JavaStreamingContext 的socketTextStream 方法可以创建一个数据源为socket 的网路端口的数据源 val blackList=Array(("tom",true)) val blackListRDD=jssc.sparkContext.parallelize(blackList,5); val lines = jssc.socketTextStream("www.codeguoj.cn", 9999) val userAdsClickLogDStream=lines.map(adfLog=>(adfLog.split(" ")(1),adfLog)) val validAdfClickRDD=userAdsClickLogDStream.transform(userAdsClickLogRDD=>{ val joinedRDD=userAdsClickLogRDD.leftOuterJoin(blackListRDD) val fliterRDD=joinedRDD.filter(tupe=>{ if(tupe._2._2.getOrElse(false)){ false }else{ true } }) val clickedLogRDD=fliterRDD.map(tupe=>tupe._2._1) clickedLogRDD }) validAdfClickRDD.print() jssc.start() jssc.awaitTermination() } }