十年网站开发经验 + 多家企业客户 + 靠谱的建站团队
量身定制 + 运营维护+专业推广+无忧售后,网站问题一站解决
Spark Streaming是Spark核心API的扩展,可以实现可伸缩、高吞吐量、具备容错机制的实时流时数据的处理。支持多种数据源,比如Kafka、Flume、Twitter、ZeroMQ、Kinesis 以及TCP sockets。
网站前端页面设计会用DIV+CSS架构,布局出来的网站外观简洁大气。HTML静态,html5+CSS3网站,自适应电脑、手机、平板,符合用户体验的习惯,更容易与用户产生互动。专业网络公司的服务理念是“高性价比建站,让企业网站具备营销价值,促进长期合作共赢模式”。可以使用诸如map、reduce、join和window等高级函数进行复杂算法(比如,机器学习和图计算)的处理。最后还可以将处理结果存储到文件系统,数据库和仪表盘。
Spark Streaming接收实时流的数据,并根据一定的时间间隔拆分成一批批的数据,然后通过Spark Engine处理这些批数据,最终得到处理后的一批批结果数据。
Spark Streaming提供了一个叫做DStream(discretized stream,离散流)的抽象概念,DStream由一系列的RDD组成,表示每个批次中连续的数据流。DStream可以从输入源(比如,Kafka、Flume、Kinesis等)中创建,也可以从其他DStream中使用高级算子操作转换生成。
DStream的所有操作其实都是对DStream中所有RDD的操作。比如,在单词统计案例中,flatMap转化操作会应用到每个行RDD上来生成单词RDD。
Receiver:Spark Streaming内置的数据流接收器或自定义接收器,用于从数据源接收源源不断的数据流。
CurrentBuffer:用于缓存输入流接收器接收的数据流。
BlockIntervalTimer:一个定时器,用于将CurrentBuffer中缓存的数据流封装为Block后放入blocksForPushing队列中。
BlocksForPushing:待处理的Block
BlockPushingThread:此线程每隔100毫秒从BlocksForPushing队列中取出一个Block存入存储系统,并缓存到ReceivedBlockQueue队列中。
Block Batch:Block批次,按照批次时间间隔,从ReceivedBlockQueue队列中获取一批Block。
DStream转化操作分为无状态(stateless)和有状态(stateful)两种。
无状态转化操作中,每个批次的处理不依赖于之前批次的数据。
无状态转化操作就是把简单的RDD转化操作应用到每个批次上,转化DStream中的每个RDD。
常用的无状态转化操作
函数名称 | 作用 | scala示例 |
---|---|---|
map() | 对DStream中的每个元素应用指定函数,返回由各元素输出的元素组成的DStream | ds.map(x => x+1) |
flatMap() | 对DStream中的每个元素应用指定函数,返回由各元素输出的迭代器组成的DStream | ds.flatMap(x => x.split(" ")) |
filter | 返回由给定DStream中通过筛选的元素组成的DStream | ds.filter(x => x!=1) |
repartition() | 改变DStream的分区数 | ds.repartition(10) |
reduceByKey | 将每个批次中键相同的记录聚合 | ds.reduceByKey((x,y) => x+y) |
groupByKey | 将每个批次中的记录根据键分组 | ds.groupByKey() |
使用map()和reduceByKey()在每个时间区间中对日志根据IP地址进行计数。
//假设ApacheAccessingLog是用来从Apache日志中解析条目的工具类
val accessLogDStream = logData.map(line => ApacheAccessingLog.parseFromLogLine(line))
val ipDStream = accessLogsDStream.map(entry => (entry.getIpAddress(), 1)
val ipCountsDStream = ipDStream.reduceByKey((x,y) => x+y)
//假设ApacheAccessingLog是用来从Apache日志中解析条目的工具类
static final class IpTuple implements PairFunction {
public Tuple2 call(ApacheAccessLog log) {
return new Tuple2<>(log.getIpAddress(), 1L);
}
}
JavaDStream accessLogDStream = logData.map(new ParseFromLogLine());
JavaPairDStream ipDStream = accessLogDStream.mapToPair(new IpTuple());
JavaPairDStream(String, Long) ipCountsDStream = ipDStream.reduceByKey(new LongSumReducer());
以IP地址为键,将请求计数的数据和传输数据量的数据连接起来
val ipBytesDStream = accessLogsDStream.map(entry => (entry.getIpAddress(), entry.getContentSize()))
val ipBytesSumDStream = ipBytesDStream.reduceByKey((x,y) => x+y)
val ipBytesRequestCountDStream = ipCountsDStream.join(ipBytesSumDStream)
JavaPairDStream ipBytesDStream = accessLogsDStream.mapToPair(new IpContentTuple());
JavaPairDStream ipBytesSumDStream = ipBytesDStream.reduceByKey(new LongSumReducer());
JavaPairDStream> ipBytesRequestCountDStream = ipCountsDStream.join(ipBytesSumDStream);
使用transform()操作实现自定义转化操作,从日志记录中提取异常值。
val outlierDStream = accessLogsDStream.transform{
rdd => extractOutliers(rdd)
}
JavaPairDStream ipRawDStream = accessLogsDStream.transform(
new Function, JavaRDD>() {
public JavaPairRDD call(JavaRDD rdd) {
return extractOutliers(rdd);
}
}
);
DStream的有状态转化操作是跨时间区间跟踪数据的操作,先前批次的数据也被用来在新的批次中计算结果。
有状态转化操作主要有两种类型:滑动窗口和updateStateByKey()。前者以一个时间阶段为滑动窗口进行操作,后者用来跟踪每个键的状态变化。
有状态转化操作需要在StreamingContext中打开检查点机制确保容错性。
ssc.checkpoint("hdfs://...")
基于窗口的操作会在一个比StreamingContext批次间隔更长的时间范围内,通过整合多个批次的结果,计算出整个窗口的结果。
基于窗口的转化操作需要两个参数,分别是窗口时长和滑动时长。两者都是批次间隔的整数倍。
窗口时长:控制每次计算最近的windowDuration/batchInterval个批次的数据。
使用window()对窗口进行计数
val accessLogsWindow = accessLogsDStream.window(Seconds(30), Seconds(10))
val windowCounts = accessLogsWindow.count()
JavaDStream accessLogsWindow = accessLogsDStream.window(Durations.seconds(30), Duration.seconds(10));
JavaDStream windowCounts = accessLogsWindow.count();
使用reduceByKeyAndWindow对每个IP地址的访问量计数
val ipDStream = accessLogsDStream.map(logEntry => (logEntry.getIpAddress(), 1))
val ipCountDStream = ipDStream.reduceByKeyAndWindow(
{(x,y) => x+y}, //加入新进入窗口的批次中的元素
{(x,y) => x-y}, //移除离开窗口的老批次中的元素
Seconds(30), //窗口时长
Seconds(10) //滑动步长
)
class ExtractIp extends PairFunction {
public Tuple2 call(ApacheAccessLog entry) {
return new Tuple2(entry.getIpAddress(), 1L);
}
}
class AddLongs extends Function2() {
public Long call(Long v1, Long v2) {
return v1 + v2;
}
}
class SubtractLongs extends Function2() {
public Long call(Long v1, Long v2) {
return v1 - v2;
}
}
JavaPairDStream ipAddressPairDStream = accessLogsDStream.mapToPair(new ExtractIp());
JavaPairDStream ipCountDStream = ipAddressPairDStream.reduceByKeyAndWindow(
new AddLongs(), //加上新进入窗口的批次中的元素
new SubtractLongs(), //移除离开窗口的老批次中的元素
Durations.seconds(30), //窗口时长
Durations.seconds(10) //滑动步长
)
使用countByWindow和countByValueAndWindow对窗口计数
scala
val ipDStream = accessLogsDStream.map{entry => entry.getIpAddress()}
val ipAddre***equestCount = ipDStream.countByValueAndWindow(Seconds(30), Seconds(10))
val requestCount = accessLogsDStream.countByWindow(Seconds(30), Seconds(10))
JavaDStream ip = accessLogsDStream.map(new Function() {
public String call(ApacheAccessLog entry) {
return entry.getIpAddress();
}
});
JavaDStream requestCount = accessLogsDStream.countByWindow(Dirations.seconds(30), Durations.seconds(10));
JavaPairDStream ipAddre***equestCount = ip.countByValueAndWindow(Dirations.seconds(30), Durations.seconds(10));
updateStateByKey提供了跨批次维护状态的功能,用于键值对形式的DStream。
updateStateByKey提供了一个update(events, oldState)函数,接收与某键相关的事件及该键之前对应的状态,返回该键对应的新状态。
使用updateStateByKey()跟踪日志消息中各HTTP响应代码的计数。
def updateRunningSum(values: Seq[Long], state: Option[Long]) = {
Some(state.getOrElse(0L) + values.size)
}
val responseCodeDStream = accessLogsDStream.map(log => (log.getResponseCode(), 1L))
val responseCodeCountDStream = responseCodeDStream.updateStateByKey(updateRunningSum _)
class UpdateRunningSum implements Function2, Optional, Optional> {
public Optional call(List nums, Optional current) {
long sum = current.or(0L);
return Optional.of(sum + nums.size());
}
};
JavaPairDStream responseCodeCountDStream = accessLogsDStream.mapToPair(
new PairFunction() {
public Tuple2 call(ApacheAccessLog log) {
return new Tuple2(log.getResponseCode(), 1L);
}
}
).updateStateByKey(new UpdateRunningSum());
DStream行动操作同RDD的行动操作。比如,将DStream保存为SequenceFile文件。
val writableIpAddre***equestCount = ipAddre***equestCount.map{
(ip, count) =>
JavaPairDStream writableDStream = ipDStream.mapToPair(
new PairFunction, Text, LongWritable>() {
public Tuple2 call(Tuple2 e) {
return new Tuple2(new Text(e._1()), new LongWritable(e._2()));
}
}
);
writableDStream.saveAsHadoopFiles("outputDir", "txt", Text.class, LongWritable.class, SequenceFileOutputFormat.class);
忠于技术,热爱分享。欢迎关注公众号:java大数据编程,了解更多技术内容。
另外有需要云服务器可以了解下创新互联scvps.cn,海内外云服务器15元起步,三天无理由+7*72小时售后在线,公司持有idc许可证,提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案,具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势,专为企业上云打造定制,能够满足用户丰富、多元化的应用场景需求。