这是flink入门的一个小项目,分享出来一起学习,路过就给个赞呗~~~~~~
words count 简而言之就是统计单词出现的次数,项目分成两部分:1、数据来源于 txt 文件, 2、通过socket在线发送数据,模拟实时系统。
开发环境:
IDEA + JDK1.8 + MAVEN
二、 txt 文件读取数据并统计 输出org.apache.flink flink-java1.2.0 org.apache.flink flink-streaming-java_2.101.2.0 org.apache.flink flink-clients_2.101.2.0
话不多说,直接上代码
package com.samur.flink.test2; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.operators.AggregateOperator; import org.apache.flink.api.java.operators.DataSource; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.util.Collector; public class WordsCount { public static void main(String[] args) throws Exception { // 1、获取开发环境 ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); // 2、读取文件数据 DataSource三、通过socket 发送数据,然后实时监听统计source = env.readTextFile("D:\myself\project\WC\src\main\resources\HW.txt"); // 3、map *** 作,将数据打散,然后组成元组 AggregateOperator > out = source.flatMap(new FlatMapFunction >() { @Override public void flatMap(String s, Collector > collector) throws Exception { String[] words = s.split(" "); for (String word : words) { collector.collect(new Tuple2<>(word, 1)); } } }) .groupBy(0) // 根据元组 的第一位(也就是word) 数据进行 group 聚合 .sum(1); // sum 元组的第二位 out.print(); // 打印输出 } }
1、开发前准备:
a、下载安装netcat(下载地址:https://eternallybored.org/misc/netcat/ );
b、在windows电脑上,win + R,然后输入 cmd 启动小黑窗,输入 nc -l -p 7777 回车;
2、在IDEA中编写执行如下代码
package com.samur.flink.test2; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.util.Collector; public class WordsCountStream { public static void main(String[] args) throws Exception { // 获取开发环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 创建 socket 连接 DataStreamSourcesource = env.socketTextStream("localhost", 7777); // 3、map *** 作,将数据打散,然后组成元组 DataStream > out = source.flatMap(new FlatMapFunction >() { @Override public void flatMap(String s, Collector > collector) throws Exception { String[] words = s.split(" "); for (String word : words) { collector.collect(new Tuple2(word, 1)); } } }) .keyBy(0) // 根据元组 的第一位(也就是word) 数据进行 group 聚合 .sum(1); // sum 元组的第二位 out.print(); // 打印输出 env.execute(); // 启动执行 } }
在第2 步的小黑窗中 输入字符串即可。
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)