13.sparkstreaming之快速入门-创新互联

简介

Spark Streaming是Spark核心API的扩展,可以实现可伸缩、高吞吐量、具备容错机制的实时流时数据的处理。支持多种数据源,比如Kafka、Flume、Twitter、ZeroMQ、Kinesis 以及TCP sockets。

网站前端页面设计会用DIV+CSS架构,布局出来的网站外观简洁大气。HTML静态,html5+CSS3网站,自适应电脑、手机、平板,符合用户体验的习惯,更容易与用户产生互动。专业网络公司的服务理念是“高性价比建站,让企业网站具备营销价值,促进长期合作共赢模式”。

可以使用诸如map、reduce、join和window等高级函数进行复杂算法(比如,机器学习和图计算)的处理。最后还可以将处理结果存储到文件系统,数据库和仪表盘。

13.spark streaming之快速入门

架构与抽象

抽象

Spark Streaming接收实时流的数据,并根据一定的时间间隔拆分成一批批的数据,然后通过Spark Engine处理这些批数据,最终得到处理后的一批批结果数据。

13.spark streaming之快速入门

Spark Streaming提供了一个叫做DStream(discretized stream,离散流)的抽象概念,DStream由一系列的RDD组成,表示每个批次中连续的数据流。DStream可以从输入源(比如,Kafka、Flume、Kinesis等)中创建,也可以从其他DStream中使用高级算子操作转换生成。

13.spark streaming之快速入门

DStream的所有操作其实都是对DStream中所有RDD的操作。比如,在单词统计案例中,flatMap转化操作会应用到每个行RDD上来生成单词RDD。

13.spark streaming之快速入门

架构

13.spark streaming之快速入门

  • Receiver:Spark Streaming内置的数据流接收器或自定义接收器,用于从数据源接收源源不断的数据流。

  • CurrentBuffer:用于缓存输入流接收器接收的数据流。

  • BlockIntervalTimer:一个定时器,用于将CurrentBuffer中缓存的数据流封装为Block后放入blocksForPushing队列中。

  • BlocksForPushing:待处理的Block

  • BlockPushingThread:此线程每隔100毫秒从BlocksForPushing队列中取出一个Block存入存储系统,并缓存到ReceivedBlockQueue队列中。

  • Block Batch:Block批次,按照批次时间间隔,从ReceivedBlockQueue队列中获取一批Block。

  • JobGenerator:Job生成器,用于给每一批Block生成一个Job。

DStream 转化操作

DStream转化操作分为无状态(stateless)和有状态(stateful)两种。

  • 无状态转化操作中,每个批次的处理不依赖于之前批次的数据。

  • 有状态转化操作需要使用之前批次的数据或中间结果来计算当前批次的数据。

无状态转化操作

无状态转化操作就是把简单的RDD转化操作应用到每个批次上,转化DStream中的每个RDD。

常用的无状态转化操作

函数名称 作用 scala示例
map()对DStream中的每个元素应用指定函数,返回由各元素输出的元素组成的DStreamds.map(x => x+1)
flatMap()对DStream中的每个元素应用指定函数,返回由各元素输出的迭代器组成的DStreamds.flatMap(x => x.split(" "))
filter返回由给定DStream中通过筛选的元素组成的DStreamds.filter(x => x!=1)
repartition()改变DStream的分区数ds.repartition(10)
reduceByKey将每个批次中键相同的记录聚合ds.reduceByKey((x,y) => x+y)
groupByKey将每个批次中的记录根据键分组ds.groupByKey()
  • 使用map()和reduceByKey()在每个时间区间中对日志根据IP地址进行计数。

    • scala
    //假设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)
    • java
    //假设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地址为键,将请求计数的数据和传输数据量的数据连接起来

    • scala
    val ipBytesDStream = accessLogsDStream.map(entry => (entry.getIpAddress(), entry.getContentSize()))
    val ipBytesSumDStream = ipBytesDStream.reduceByKey((x,y) => x+y)
    val ipBytesRequestCountDStream = ipCountsDStream.join(ipBytesSumDStream)
    • java
    JavaPairDStream ipBytesDStream = accessLogsDStream.mapToPair(new IpContentTuple());
    JavaPairDStream ipBytesSumDStream = ipBytesDStream.reduceByKey(new LongSumReducer());
    JavaPairDStream> ipBytesRequestCountDStream = ipCountsDStream.join(ipBytesSumDStream);
  • 使用transform()操作实现自定义转化操作,从日志记录中提取异常值。

    • scala
    val outlierDStream = accessLogsDStream.transform{
        rdd => extractOutliers(rdd)
    }
    • java
    JavaPairDStream ipRawDStream = accessLogsDStream.transform(
        new Function, JavaRDD>() {
            public JavaPairRDD call(JavaRDD rdd) {
                return extractOutliers(rdd);
            }
        }
    );

有状态转化操作

DStream的有状态转化操作是跨时间区间跟踪数据的操作,先前批次的数据也被用来在新的批次中计算结果。

有状态转化操作主要有两种类型:滑动窗口和updateStateByKey()。前者以一个时间阶段为滑动窗口进行操作,后者用来跟踪每个键的状态变化。

设置检查点

有状态转化操作需要在StreamingContext中打开检查点机制确保容错性。

ssc.checkpoint("hdfs://...")
基于窗口的转化操作
简介

基于窗口的操作会在一个比StreamingContext批次间隔更长的时间范围内,通过整合多个批次的结果,计算出整个窗口的结果。

基于窗口的转化操作需要两个参数,分别是窗口时长和滑动时长。两者都是批次间隔的整数倍。

  • 窗口时长:控制每次计算最近的windowDuration/batchInterval个批次的数据。

  • 滑动步长:默认值与批次间隔相等。用来控制对新DStream进行计算的时间间隔。
简单案例
  • 使用window()对窗口进行计数

    • scala
    val accessLogsWindow = accessLogsDStream.window(Seconds(30), Seconds(10))
    val windowCounts = accessLogsWindow.count()
    • java
    JavaDStream accessLogsWindow = accessLogsDStream.window(Durations.seconds(30), Duration.seconds(10));
    JavaDStream windowCounts = accessLogsWindow.count();
  • 使用reduceByKeyAndWindow对每个IP地址的访问量计数

    • scala
    val ipDStream = accessLogsDStream.map(logEntry => (logEntry.getIpAddress(), 1))
    val ipCountDStream = ipDStream.reduceByKeyAndWindow(
        {(x,y) => x+y}, //加入新进入窗口的批次中的元素
        {(x,y) => x-y}, //移除离开窗口的老批次中的元素
        Seconds(30), //窗口时长
        Seconds(10) //滑动步长
    )
    • java
    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))
    • java
    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转化操作
简介

updateStateByKey提供了跨批次维护状态的功能,用于键值对形式的DStream。

updateStateByKey提供了一个update(events, oldState)函数,接收与某键相关的事件及该键之前对应的状态,返回该键对应的新状态。

  • events:当前批次中收到的事件列表
  • oldState:一个可选的状态对象,存放在Option内;如果一个键没有之前的状态,这个值为空。
  • newState:由函数返回,也以Option形式存在;可以返回一个空的Option表示删除该状态。
简单案例

使用updateStateByKey()跟踪日志消息中各HTTP响应代码的计数。

  • scala
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 _)
  • java
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 行动操作

DStream行动操作同RDD的行动操作。比如,将DStream保存为SequenceFile文件。

  • scala
val writableIpAddre***equestCount = ipAddre***equestCount.map{
    (ip, count) => 
  • java
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大数据编程,了解更多技术内容。

13.spark streaming之快速入门

另外有需要云服务器可以了解下创新互联scvps.cn,海内外云服务器15元起步,三天无理由+7*72小时售后在线,公司持有idc许可证,提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案,具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势,专为企业上云打造定制,能够满足用户丰富、多元化的应用场景需求。


网页标题:13.sparkstreaming之快速入门-创新互联
地址分享:http://pwwzsj.com/article/epggs.html