Flink入门：读取Kafka实时数据流，实现WordCount_flink读取kafka数据进行统计资源-CSDN文库

版权申诉

5星 · 超过95%的资源 125 浏览量 2021-01-07 11:35:51 上传评论收藏 437KB PDF 举报

资源推荐

资源详情

资源评论

Flink入门：读取入门：读取Kafka实时数据流，实现实时数据流，实现WordCount

本文主要介绍Flink接收一个Kafka文本数据流，进行WordCount词频统计，然后输出到标准输出上。通过本文你可以了解如何编写和运行Flink程序。

代码拆解代码拆解

首先要设置Flink的执行环境：

// 创建Flink执行环境

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

设置Kafka相关参数，连接对应的服务器和端口号，读取名为Shakespeare的Topic中的数据源，将数据源命名为stream：

// Kafka参数

Properties properties = new Properties();

properties.setProperty("bootstrap.servers", "localhost:9092");

properties.setProperty("group.id", "flink-group");

String inputTopic = "Shakespeare";

String outputTopic = "WordCount";

// Source

FlinkKafkaConsumer consumer =

new FlinkKafkaConsumer(inputTopic, new SimpleStringSchema(), properties);

DataStream stream = env.addSource(consumer);

使用Flink算子处理这个数据流：

// Transformations

// 使用Flink算子对输入流的文本进行操作

// 按空格切词、计数、分区、设置时间窗口、聚合

DataStream<Tuple2> wordCount = stream

.flatMap((String line, Collector<Tuple2> collector) -> {

String[] tokens = line.split("\s");

// 输出结果 (word, 1)

for (String token : tokens) {

if (token.length() > 0) {

collector.collect(new Tuple2(token, 1));

}

}

})

.returns(Types.TUPLE(Types.STRING, Types.INT))

.keyBy(0)

.timeWindow(Time.seconds(5))

.sum(1);

这里使用的是Flink提供的DataStream级别的API，主要包括转换、分组、窗口和聚合等操作。

将数据流打印：

// Sink

wordCount.print();

最后执行这个程序：

// execute

env.execute("kafka streaming word count");

env.execute 是启动Flink作业所必需的，只有在execute()被调用时，之前调用的各个操作才会在提交到集群上或本地计算机上执行。

完整代码如下：

import org.apache.flink.api.common.serialization.SimpleStringSchema;

import org.apache.flink.api.common.typeinfo.Types;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.datastream.DataStream;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import org.apache.flink.util.Collector;

import java.util.Properties;

public class WordCountKafkaInStdOut {

public static void main(String[] args) throws Exception {

// 创建Flink执行环境

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// Kafka参数

Properties properties = new Properties();

本内容试读结束，登录后可阅读更多

下载后可阅读完整内容，剩余3页未读，立即下载

内容反馈

版权申诉

无用书生~

2023-01-09

内容与描述一致，超赞的资源，值得借鉴的内容很多，支持！
qq_53228537

2023-05-12

感谢大佬分享的资源，对我启发很大，给了我新的灵感。
m0_65499649

2024-08-20

资源很受用，资源主总结的很全面，内容与描述一致，解决了我当下的问题。
2301_76831817

2023-08-03

感谢资源主分享的资源解决了我当下的问题，非常有用的资源。

weixin_38659527

粉丝: 6
资源: 871

最新资源

资源上传下载、课程学习等过程中有任何疑问或建议，欢迎提出宝贵意见哦~我们会及时处理！点击此处反馈

feedback-tip