最简单流处理引擎——Kafka Streams简介

kafka在0.10.0.0版本以前的定位是分布式，分区化的，带备份机制的日志提交服务。而kafka在这之前也没有提供数据处理的顾服务。大家的流处理计算主要是还是依赖于storm，spark streaming，flink等流式处理框架。

storm，spark streaming，flink流处理的三驾马车各有各的优势.

storm低延迟，并且在市场中占有一定的地位，目前很多公司仍在使用。

spark streaming借助spark的体系优势，活跃的社区，也占有一定的份额。

而flink在设计上更贴近流处理，并且有便捷的api，未来一定很有发展。

但是他们都离不开kafka的消息中转，所以kafka于0.10.0.0版本推出了自己的流处理框架，kafka streams。kafka的定位也正式成为apache kafka® is a distributed streaming platform，分布式流处理平台。

实时流式计算

近几年来实时流式计算发展迅速，主要原因是实时数据的价值和对于数据处理架构体系的影响。实时流式计算包含了 无界数据近实时一致性可重复结果 等等特征。a type of data processing engine that is designed with infinite data sets in mind 一种考虑了无线数据集的数据处理引擎。

1、无限数据：一种不断增长的，基本上无限的数据集。这些通常被称为“流式数据”。无限的流式数据集可以称为无界数据，相对而言有限的批量数据就是有界数据。

2、无界数据处理：一种持续的数据处理模式，应用于上面的无界数据。批量处理数据（离线计算）也可以重复运行来处理数据，但是会有性能的瓶颈。

3、低延迟，近实时的结果：相对于离线计算而言，离线计算并没有考虑延迟的问题。

解决了两个问题，流处理可以提代批处理系统：

1、正确性：有了这个，就和批量计算等价了。

streaming需要能随着时间的推移依然能计算一定时间窗口的数据。spark streaming通过微批的思想解决了这个问题，实时与离线系统进行了一致性的存储，这一点在未来的实时计算系统中都应该满足。

2、推理时间的工具：这可以让我们超越批量计算。

好的时间推理工具对于处理不同事件的无界无序数据至关重要。

而时间又分为事件时间和处理时间。

还有很多实时流式计算的相关概念，这里不做赘述。

kafka streams简介

kafka streams被认为是开发实时应用程序的最简单方法。它是一个kafka的客户端api库，编写简单的java和scala代码就可以实现流式处理。

优势：

弹性，高度可扩展，容错
部署到容器，vm，裸机，云
同样适用于小型，中型和大型用例
与kafka安全性完全集成
编写标准java和scala应用程序
在mac，linux，windows上开发
exactly-once 语义

用例：

纽约时报使用apache kafka和kafka streams将发布的内容实时存储和分发到各种应用程序和系统，以供读者使用。

pinterest大规模使用apache kafka和kafka streams来支持其广告基础架构的实时预测预算系统。使用kafka streams，预测比以往更准确。

作为欧洲领先的在线时尚零售商，zalando使用kafka作为esb（企业服务总线），帮助我们从单一服务架构转变为微服务架构。使用kafka处理使我们的技术团队能够实现近乎实时的商业智能。

荷兰合作银行是荷兰三大银行之一。它的数字神经系统business event bus由apache kafka提供支持。它被越来越多的财务流程和服务所使用，其中之一就是rabo alerts。此服务会在财务事件时实时向客户发出警报，并使用kafka streams构建。

line使用apache kafka作为我们服务的中央数据库，以便彼此通信。每天产生数亿亿条消息，用于执行各种业务逻辑，威胁检测，搜索索引和数据分析。line利用kafka streams可靠地转换和过滤主题，使消费者可以有效消费的子主题，同时由于其复杂而简单的代码库，保持易于维护性。

topology

kafka streams通过一个或多个拓扑定义其计算逻辑，其中拓扑是通过流（边缘）和流处理器（节点）构成的图。

拓扑中有两种特殊的处理器

源处理器：源处理器是一种特殊类型的流处理器，没有任何上游处理器。它通过使用来自这些主题的记录并将它们转发到其下游处理器，从一个或多个kafka主题为其拓扑生成输入流。
接收器处理器：接收器处理器是一种特殊类型的流处理器，没有下游处理器。它将从其上游处理器接收的任何记录发送到指定的kafka主题。

在正常处理器节点中，还可以把数据发给远程系统。因此，处理后的结果可以流式传输回kafka或写入外部系统。

kafka在这当中提供了最常用的数据转换操作，例如map，filter，join和aggregations等，简单易用。

当然还有一些关于时间，窗口，聚合，乱序处理等。未来再一一做详细介绍，下面我们进行简单的入门案例开发。

快速入门

首先提供wordcount的java版和scala版本。

java8+：

import org.apache.kafka.common.serialization.serdes;
import org.apache.kafka.common.utils.bytes;
import org.apache.kafka.streams.kafkastreams;
import org.apache.kafka.streams.streamsbuilder;
import org.apache.kafka.streams.streamsconfig;
import org.apache.kafka.streams.kstream.kstream;
import org.apache.kafka.streams.kstream.ktable;
import org.apache.kafka.streams.kstream.materialized;
import org.apache.kafka.streams.kstream.produced;
import org.apache.kafka.streams.state.keyvaluestore;
 
import java.util.arrays;
import java.util.properties;
 
public class wordcountapplication {
 
    public static void main(final string[] args) throws exception {
        properties props = new properties();
        props.put(streamsconfig.application_id_config, "wordcount-application");
        props.put(streamsconfig.bootstrap_servers_config, "kafka-broker1:9092");
        props.put(streamsconfig.default_key_serde_class_config, serdes.string().getclass());
        props.put(streamsconfig.default_value_serde_class_config, serdes.string().getclass());
 
        streamsbuilder builder = new streamsbuilder();
        kstream<string, string> textlines = builder.stream("textlinestopic");
        ktable<string, long> wordcounts = textlines
            .flatmapvalues(textline -> arrays.aslist(textline.tolowercase().split("\\w+")))
            .groupby((key, word) -> word)
            .count(materialized.<string, long, keyvaluestore<bytes, byte[]>>as("counts-store"));
        wordcounts.tostream().to("wordswithcountstopic", produced.with(serdes.string(), serdes.long()));
 
        kafkastreams streams = new kafkastreams(builder.build(), props);
        streams.start();
    }
 
}

scala：

import java.util.properties
import java.util.concurrent.timeunit
 
import org.apache.kafka.streams.kstream.materialized
import org.apache.kafka.streams.scala.implicitconversions._
import org.apache.kafka.streams.scala._
import org.apache.kafka.streams.scala.kstream._
import org.apache.kafka.streams.{kafkastreams, streamsconfig}
 
object wordcountapplication extends app {
  import serdes._
 
  val props: properties = {
    val p = new properties()
    p.put(streamsconfig.application_id_config, "wordcount-application")
    p.put(streamsconfig.bootstrap_servers_config, "kafka-broker1:9092")
    p
  }
 
  val builder: streamsbuilder = new streamsbuilder
  val textlines: kstream[string, string] = builder.stream[string, string]("textlinestopic")
  val wordcounts: ktable[string, long] = textlines
    .flatmapvalues(textline => textline.tolowercase.split("\\w+"))
    .groupby((_, word) => word)
    .count()(materialized.as("counts-store"))
  wordcounts.tostream.to("wordswithcountstopic")
 
  val streams: kafkastreams = new kafkastreams(builder.build(), props)
  streams.start()
 
  sys.shutdownhookthread {
     streams.close(10, timeunit.seconds)
  }
}

如果kafka已经启动了，可以跳过前两步。

1、下载

2.3.0版本并解压缩它。请注意，有多个可下载的scala版本，我们选择使用推荐的版本（2.12）：

> tar -xzf kafka_2.12-2.3.0.tgz
> cd kafka_2.12-2.3.0

2、启动

kafka使用zookeeper，因此如果您还没有zookeeper服务器，则需要先启动它。

> bin/zookeeper-server-start.sh config/zookeeper.properties
[2013-04-22 15:01:37,495] info reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.quorumpeerconfig)
...

启动kafka服务器：

> bin/kafka-server-start.sh config/server.properties
[2013-04-22 15:01:47,028] info verifying properties (kafka.utils.verifiableproperties)
[2013-04-22 15:01:47,051] info property socket.send.buffer.bytes is overridden to 1048576 (kafka.utils.verifiableproperties)
...

3、创建topic 启动生产者

我们创建名为streams-plaintext-input的输入主题和名为streams-wordcount-output的输出主题：

> bin/kafka-topics.sh --create \
    --bootstrap-server localhost:9092 \
    --replication-factor 1 \
    --partitions 1 \
    --topic streams-plaintext-input
created topic "streams-plaintext-input".


> bin/kafka-topics.sh --create \
    --bootstrap-server localhost:9092 \
    --replication-factor 1 \
    --partitions 1 \
    --topic streams-wordcount-output \
    --config cleanup.policy=compact
created topic "streams-wordcount-output".

查看：

> bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe
 
topic:streams-plaintext-input   partitioncount:1    replicationfactor:1 configs:
    topic: streams-plaintext-input  partition: 0    leader: 0   replicas: 0 isr: 0
topic:streams-wordcount-output  partitioncount:1    replicationfactor:1 configs:cleanup.policy=compact
    topic: streams-wordcount-output partition: 0    leader: 0   replicas: 0 isr: 0

4、启动wordcount

以下命令启动wordcount演示应用程序：

> bin/kafka-run-class.sh org.apache.kafka.streams.examples.wordcount.wordcountdemo

演示应用程序将从输入主题stream-plaintext-input读取，对每个读取消息执行wordcount算法的计算，并连续将其当前结果写入输出主题streams-wordcount-output。因此，除了日志条目之外不会有任何stdout输出，因为结果会写回kafka。

现在我们可以在一个单独的终端中启动控制台生成器，为这个主题写一些输入数据：

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic streams-plaintext-input

并通过在单独的终端中使用控制台使用者读取其输出主题来检查wordcount演示应用程序的输出：

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
    --topic streams-wordcount-output \
    --from-beginning \
    --formatter kafka.tools.defaultmessageformatter \
    --property print.key=true \
    --property print.value=true \
    --property key.deserializer=org.apache.kafka.common.serialization.stringdeserializer \
    --property value.deserializer=org.apache.kafka.common.serialization.longdeserializer

5、处理数据

我们在生产者端输入一些数据。

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic streams-plaintext-input
all streams lead to kafka

输出端：

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
    --topic streams-wordcount-output \
    --from-beginning \
    --formatter kafka.tools.defaultmessageformatter \
    --property print.key=true \
    --property print.value=true \
    --property key.deserializer=org.apache.kafka.common.serialization.stringdeserializer \
    --property value.deserializer=org.apache.kafka.common.serialization.longdeserializer
 
all     1
streams 1
lead    1
to      1
kafka   1

继续输入：

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic streams-plaintext-input
all streams lead to kafka
hello kafka streams

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
    --topic streams-wordcount-output \
    --from-beginning \
    --formatter kafka.tools.defaultmessageformatter \
    --property print.key=true \
    --property print.value=true \
    --property key.deserializer=org.apache.kafka.common.serialization.stringdeserializer \
    --property value.deserializer=org.apache.kafka.common.serialization.longdeserializer
 
all     1
streams 1
lead    1
to      1
kafka   1
hello   1
kafka   2
streams 2

我们看到随着数据实时输入，wordcount的结果实时的输出了。

6、停止程序

您现在可以通过ctrl-c按顺序停止控制台使用者，控制台生产者，wordcount应用程序，kafka代理和zookeeper服务器。

什么是kafka？
kafka监控工具汇总
kafka快速入门
kafka核心之consumer
kafka核心之producer

替代flume——kafka connect简介

更多实时计算,flink,kafka等相关技术博文，欢迎关注实时流式计算

最简单流处理引擎——Kafka Streams简介

实时流式计算

kafka streams简介

优势：

用例：

topology

快速入门

1、下载

2、启动

3、创建topic 启动生产者

4、启动wordcount

5、处理数据

6、停止程序

相关推荐

spark怎么读取kafka数据

在微服务架构中怎么使用Kafka

Kafka中的事务管理方法是什么

怎么监控和管理Kafka集群

Kafka在金融行业中如何应用

Kafka中Partition和Rebalance机制是什么

Kafka如何实现网站实时访问日志处理

Kafka消费者组和负载均衡策略是什么