kafkaStream处理实时流式计算

目录

1 实时流式计算

1.1 概念

一般流式计算会与批量计算相比较。在流式计算模型中,输入是持续的,可以认为在时间上是无界的,也就意味着,永远拿不到全量数据去做计算。同时,计算结果是持续输出的,也即计算结果在时间上也是无界的。流式计算一般对实时性要求较高,同时一般是先定义目标计算,然后数据到来之后将计算逻辑应用于数据。同时为了提高计算效率,往往尽可能采用增量计算代替全量计算。

《kafkaStream处理实时流式计算》

流式计算就相当于上图的右侧扶梯,是可以源源不断的产生数据,源源不断的接收数据,没有边界。

1.2 应用场景

  • 日志分析

网站的用户访问日志进行实时的分析,计算访问量,用户画像,留存率等等,实时的进行数据分析,帮助企业进行决策

  • 大屏看板统计

可以实时的查看网站注册数量,订单数量,购买数量,金额等。

  • 公交实时数据

可以随时更新公交车方位,计算多久到达站牌等

  • 实时文章分值计算

头条类文章的分值计算,通过用户的行为实时文章的分值,分值越高就越被推荐。

1.3 技术方案选型

  • Hadoop

  • Apche Storm

Storm 是一个分布式实时大数据处理系统,可以帮助我们方便地处理海量数据,具有高可靠、高容错、高扩展的特点。是流式框架,有很高的数据吞吐能力。

  • Kafka Stream

可以轻松地将其嵌入任何Java应用程序中,并与用户为其流应用程序所拥有的任何现有打包,部署和操作工具集成。

2 Kafka Stream

2.1 概述

Kafka Stream是Apache Kafka从0.10版本引入的一个新Feature。它是提供了对存储于Kafka内的数据进行流式处理和分析的功能。

Kafka Stream的特点如下:

  • Kafka Stream提供了一个非常简单而轻量的Library,它可以非常方便地嵌入任意Java应用中,也可以任意方式打包和部署
  • 除了Kafka外,无任何外部依赖
  • 充分利用Kafka分区机制实现水平扩展和顺序性保证
  • 通过可容错的state store实现高效的状态操作(如windowed join和aggregation)
  • 支持正好一次处理语义
  • 提供记录级的处理能力,从而实现毫秒级的低延迟
  • 支持基于事件时间的窗口操作,并且可处理晚到的数据(late arrival of records)
  • 同时提供底层的处理原语Processor(类似于Storm的spout和bolt),以及高层抽象的DSL(类似于Spark的map/group/reduce)

2.2 Kafka Streams的关键概念

(1)Stream处理拓扑

  • 是Kafka Stream提出的最重要的抽象概念:它表示一个无限的,不断更新的数据集。流是一个有序的,可重放(反复的使用),不可变的容错序列,数据记录的格式是键值对(key-value)。
  • 通过Kafka Streams编写一个或多个的计算逻辑的处理器拓扑。其中处理器拓扑是一个由流(边缘)连接的流处理(节点)的图。
  • 流处理器处理器拓扑中的一个节点;它表示一个处理的步骤,用来转换流中的数据(从拓扑中的上游处理器一次接受一个输入消息,并且随后产生一个或多个输出消息到其下游处理器中)。

(2)在拓扑中有两个特别的处理器:

  • 源处理器(Source Processor):源处理器是一个没有任何上游处理器的特殊类型的流处理器。它从一个或多个kafka主题生成输入流。通过消费这些主题的消息并将它们转发到下游处理器。

  • Sink处理器:sink处理器是一个没有下游流处理器的特殊类型的流处理器。它接收上游流处理器的消息发送到一个指定的Kafka主题

2.3 KStream&KTable

(1)数据结构类似于map,如下图,key-value键值对

(2)KStream

KStream数据流(data stream),即是一段顺序的,可以无限长,不断更新的数据集。
数据流中比较常记录的是事件,这些事件可以是一次鼠标点击(click),一次交易,或是传感器记录的位置数据。

KStream负责抽象的,就是数据流。与Kafka自身topic中的数据一样,类似日志,每一次操作都是向其中插入(insert)新数据。

为了说明这一点,让我们想象一下以下两个数据记录正在发送到流中:

(“ alice”,1)->(”alice“,3)

如果您的流处理应用是要总结每个用户的价值,它将返回4alice。为什么?因为第二条数据记录将不被视为先前记录的更新。(insert)新数据

(3)KTable

KTable传统数据库,包含了各种存储了大量状态(state)的表格。KTable负责抽象的,就是表状数据。每一次操作,都是更新插入(update)

为了说明这一点,让我们想象一下以下两个数据记录正在发送到流中:

(“ alice”,1)->(“” alice“,3)

如果您的流处理应用是要总结每个用户的价值,它将返回3alice。为什么?因为第二条数据记录将被视为先前记录的更新。

KStream – 每个新数据都包含了部分信息。

KTable – 每次更新都合并到原记录上。

2.4 Kafka Stream入门案例编写

(1)引入依赖

在之前的kafka-demo工程的pom文件中引入

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>${kafka.client.version}</version>
    <exclusions>
        <exclusion>
            <artifactId>connect-json</artifactId>
            <groupId>org.apache.kafka</groupId>
        </exclusion>
        <exclusion>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </exclusion>
    </exclusions>
</dependency>

(2)创建类

package com.oldlu.kafka.simple;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

import java.util.Arrays;
import java.util.Properties;

/** * 需求: * 接收kafka消息内容并计算消息内单词的个数 * 如: * hello kafka stareams * hello oldlu kafka * hello beijing oldlu kafka * * 结果: * hello 3 * kafka 3 * streams 1 * oldlu 2 * beijing 1 */
public class KafkaStreamFastStart {

public static void main(String[] args) {
//kafka配置信息
Properties prop = new Properties();
prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
prop.put(StreamsConfig.APPLICATION_ID_CONFIG,"streams-faststart");

//stream构建器
StreamsBuilder builder = new StreamsBuilder();

//流式计算
group(builder);

//创建kafkaStream
KafkaStreams kafkaStreams = new KafkaStreams(builder.build(),prop);

//开启kafka流计算
kafkaStreams.start();
}

/** * 实时流式计算 * @param builder */
private static void group(StreamsBuilder builder) {
//接收上游处理器的消息
KStream<String, String> stream = builder.stream("input_topic");
KStream<String, String> map = stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
/** * 把消息中的词组,转换为一个一个的单词放到集合中 * @param value * @return */
@Override
public Iterable<String> apply(String value) {
return Arrays.asList(value.split(" "));
}
}).map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
/** * 把消息的key,重新赋值,目前消息的key,就是一个个的单词,把单词作为key进行聚合 * @param key * @param value * @return */
@Override
public KeyValue<String, String> apply(String key, String value) {
return new KeyValue<>(value, value);
}
})
//根据key进行分组 目前的key 就是value,就是一个个的单词
.groupByKey()
//聚合的时间窗口 多久聚合一次
.windowedBy<sp

点赞