Linux下搭建KafkaStream架构的实践（linuxkafka）
随着大数据的迅猛发展，对于时间序列数据的处理变得越来越重要。Apache Kafka stream作为流处理核心框架，有着非常好的支持性，在大数据领域得到了广泛的应用。本文将介绍如何搭建和配置Kafka Stream架构在linux系统上运行程序，以及常见的使用方法。
### 1. 系统要求
Kafka Stream有一些关键的系统要求，如操作系统环境，使用的Java版本以及用到的Kafka Stream工具集等。搭建环境前，必须保证系统能够支持和满足Kafka Stream系统要求，才能通过后续配置步骤形成可运行程序。
### 2. 集群节点搭建
在安装集群节点之前，需要考虑集群节点数配置，确定主节点和从节点，用于分开承担不同的任务，例如主节点负责订阅消息，从节点负责处理数据。在准备环境之后，使用以下命令可完成Linux系统的Kafka Stream节点安装：
# 设置Kafka_stream_ home路径
export KAFKA_STREAMS_HOME=/usr/local/kafka_streams
# 下载安装包
wget http://download.kafka.apache.org/streams/1.5.2/kafka-streams-1.5.2-bin.tar.gz
# 解压安装包
tar -zxvf kafka-streams-1.5.2-bin.tar.gz
# 复制解压好的文件到Kafka home
mv kafka-streams-1.5.2/* $KAFKA_STREAMS_HOME
# 删除压缩文件
rm kafka-streams-1.5.2-bin.tar.gz
# 根据节点类型进行配置
# 主节点配置
# /usr/local/kafka_streams/conf/server.properties
streamConfig.broker= # 设置broker地址
# 从节点配置
# /usr/local/kafka_streams/conf/consumer.properties
bootstrap.servers= # 设置Zookeeper地址
group.id= # 设置groupid
完成集群节点的搭建之后，就可以开始利用Kafka Streams节点搭建Kafka Stream任务。
### 3. 编写Stream任务
Kafka Stream的任务形式类似于MapReduce，它可以实现从处理和聚合单词出现频度及计数等高级功能。在编写任务之前，首先需要创建Topic，使用以下命令：
# 创建主题
$KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic test-topic
# 检查主题
$KAFKA_HOME/bin/kafka-topics.sh --list --zookeeper localhost:2181
Kafka Stream的任务编写就是一个实体类，可以使用Java和Scala等编程语言编写类，内部实现Streams API：
“`Java
public class StreamExample {
public static void main(String[] args) {
// 配置文件
final Properties props = new Properties();
// 设置应用的ID
props.put(StreamsConfig.APPLICATION_ID_CONFIG, “stream-example-app”);
// 设置应用的Broker
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, “localhost:9092”);
// 设置Client ID
props.put(StreamsConfig.CLIENT_ID_CONFIG, “stream-example-client”);
// 创建StreamsBuilder
final StreamsBuilder builder = new StreamsBuilder();
// 从topic获取流
final KStream source = builder.stream(“test-topic”);
// 进行聚合
final KTable counts = stream.flatMap((key, value) ->
Arrays.asList(value.split(” “)).iterator())
.map((key, value) -> new KeyValue(value, value))
.countByKey(“counts”);
// 输出到另一个topic
counts.toStream().to(“streams-wordcount-output”);
// 创建Topology
final Topology topology = builder.build();
// 写入控制台
System.out.println(topology.describe());
//初始化一个KafkaStream对象
final KafkaStreams streams = new KafkaStreams(topology, props);
//启动程序
streams.start();
}
}
### 4. 实时流数据分析
任务编写好之后，OK！Kafka Stream的搭建以及配置和使用就完成啦。
创新互联服务器托管拥有成都T3+级标准机房资源，具备完善的安防设施、三线及BGP网络接入带宽达10T，机柜接入千兆交换机，能够有效保证服务器托管业务安全、可靠、稳定、高效运行；创新互联专注于成都服务器托管租用十余年，得到成都等地区行业客户的一致认可。
网页题目：Linux下搭建KafkaStream架构的实践（linuxkafka）
路径分享：http://www.mswzjz.cn/qtweb/news3/445953.html
攀枝花网站建设、攀枝花网站运维推广公司-贝锐智能，是专注品牌与效果的网络营销公司；服务项目有等
声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 贝锐智能