前言：
transform不是transformation，后者是对所有的转换算子的统称，transform也是transformation算子中的一个。transform算子的主要作用就是为了弥补Streaming没有提供的相关功能的操作，比如：一个DStream和RDD进行关联操作join，或者减少分区数量。

一、使用transform来完成分区减少coalsce的操作

dstream.transform(_.coalesce(newPartitionNum))

二、使用transform来完成DStream和RDD的join操作

计费系统，是电商必不可少的一个功能点。为了防止恶意的广告点击(假设商户A和B同时在某电商做了广告，A和B为竞争对手，那么如果A使用点击机器人进行对B的广告的恶意点击，那么B的广告费用将很快被用完)，必须对广告点击进行黑名单过滤。黑名单的过滤可以是ID，可以是IP等等，黑名单就是过滤的条件，利用SparkStreaming的流处理特性，可实现实时黑名单的过滤实现。可以使用leftouter join 对目标数据和黑名单数据进行关联，将命中黑名单的数据过滤掉。

?:
/*** 使用transform进行在线黑名单过滤** 使用ip作为黑名单*     27.19.74.143*     110.52.250.126* 数据的格式：*     27.19.74.143##2016-05-30 17:38:20##GET /static/image/common/faq.gif HTTP/1.1##200##1127**/
object Transform2BlacklistFileter {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("Transform2BlacklistFileter").setMaster("local[*]")
val ssc = new StreamingContext(conf, Seconds(2))
//准备黑名单val blacklist:RDD[(String, Int)] = ssc.sparkContext.parallelize(Map("27.19.74.143" -> 1,"110.52.250.126" -> 1).toList)
//网络请求，包含了黑名单中的数据val lines:DStream[String] = ssc.socketTextStream("bigdata01", 9999)//数据示例//27.19.74.143##2016-05-30 17:38:20##GET /static/image/common/faq.gif HTTP/1.1##200##1127val ip2Info:DStream[(String, String)] = lines.map(line => {val ip = line.substring(0, line.indexOf("##"))val info = line.substring(line.indexOf("##") + 2)(ip, info)})
val filtered:DStream[(String, String)] = ip2Info.transform(infoRDD => {//            infoRDD.join(blacklist)//结果是黑名单中的数据，必须要val joinedRDD:RDD[(String, (String, Option[Int]))] = infoRDD.leftOuterJoin(blacklist)
joinedRDD.filter{case (ip, (info, option)) => !option.isDefined}.map{case (ip, (info, option)) => (ip, info)}})filtered.print()ssc.start()ssc.awaitTermination()}
}

sparkStreaming算子之transform相关推荐

【SparkStreaming学习之二】 SparkStreaming算子操作
环境虚拟机:VMware 10 Linux版本:CentOS-6.5-x86_64 客户端:Xshell4 FTP:Xftp4 jdk1.8 scala-2.10.4(依赖jdk1.8) spark ...
Bookmarks(三)
Bookmarks 书签栏 tooltips提示效果,支持点击与经过显示,位置和效果可以自定义 - CSDN博客疯狂的小萝卜头 - 博客园 [Kettle从零开始]第九弹之Kettle定时任务介绍 ...
SparkStreaming DStream入门及其算子应用
什么是Dstream? 离散流(DStream)是Spark Streaming中的基本抽象,是表示相同数据流的RDD(相同类型)的连续序列. DStreams可以使用StreamingContext ...
Spark _30_SparkStreaming算子操作Driver HA
SparkStreaming算子操作 foreachRDD output operation算子,必须对抽取出来的RDD执行action类算子,代码才能执行. import org.apache.sp ...
Dstream如何应用RDD特有算子?
比如collectAsMap算子Dstream并没有,如何用呢? 可以通过foreachRDD或者transform算子间接使用. 比如 //sparkStreaming 算子测试案例wordcoun ...
SparkStreaming实时数仓——日活
文章目录一.日活需求概述思路: 二.搭建实时处理模块前期准备: 2.1 创建module 2.2 pom.xml文件中导入依赖.创建需要的package 2.3 添加需要的配置文件 2.3.1 ...
sparkStreaming:实时流数据详解
目录一.概述二.wordCount示例三.初始化StreamingContext 四.DStreams(离散数据流) 五.输入DStream和接收器 Basic sources File Str ...
spark算子大全glom_（七）Spark Streaming 算子梳理 — repartition算子
目录天小天:(一)Spark Streaming 算子梳理 - 简单介绍streaming运行逻辑天小天:(二)Spark Streaming 算子梳理 - flatMap和mapPartitio ...
大数据工程师入职京东年薪37w（附：面试真题分享）
总结:引导面试官到自己擅长的领域,掌握主动权,问题回答不一定完整,说出关键点即可. 1.项目规模,一天/月数据量,各组件版本? 数据规模:一般100M数据由300万条数据:数据量:上百G:条数:达到几 ...
Spark期末考试练习题
一.单选题 1. 下面的端口不是 Spark 自带的服务端口的是___________. A. 8080 B. 4040 C. 8090 D. 18080 2. 下面的描述不是 RDD 的特点的是__ ...

sparkStreaming算子之transform

一、使用transform来完成分区减少coalsce的操作

二、使用transform来完成DStream和RDD的join操作

sparkStreaming算子之transform相关推荐

最新文章

热门文章