哪有做网站 的长春网络营销公司
⭐简单说两句⭐
✨ 正在努力的小新~
💖 超级爱分享，分享各种有趣干货！
👩💻 提供：模拟面试 | 简历诊断 | 独家简历模板
🌈 感谢关注，关注了你就是我的超级粉丝啦！
🔒 以下内容仅对你可见~作者：后端小知识，CSDN后端领域新星创作者 |阿里云专家博主
CSDN个人主页：后端小知识
🔎GZH：
后端小知识🎉欢迎关注🔎点赞👍收藏⭐️留言📝
Flink入门学习-WordCount
我们今天来编写一个Flink入门学习案例，统计单词出现的次数
这里就先直接上手实践，先不看枯燥的理论
IDEA方式运行
我们首先创建Flink运行环境
//设置Flink运行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
然后模拟一点数据
//从集合中读取模拟数据DataStream<String> stream = env.fromElements("spark flink kafka", "spark sqoop flink", "kakfa hadoop flink");
切词做转换
stream.flatMap((String value, Collector<Tuple2<String,Integer>> out) -> {//value就是每一个元素的数据System.out.println("读取内容：" + value);//将每一个元素按照空格切分String[] split = value.split(" ");//遍历每一个单词for (String word : split) {//将每一个单词发送到下游out.collect(new Tuple2<>(word, 1));}})
返回类型
.returns(Types.TUPLE(Types.STRING, Types.INT))
keyby分组(按照tuple的第一个元素进行分组)
.keyBy(f->f.f0)
聚合统计
.sum(1);
打印结果
sum.print();
最后执行execute
env.execute();
完整代码如下
package cn.wy.chapter02;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.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;/*** @author tiancx*/
public class WordCount {public static void main(String[] args) throws Exception {//设置Flink运行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//从集合中读取模拟数据DataStream<String> stream = env.fromElements("spark flink kafka", "spark sqoop flink", "kakfa hadoop flink");SingleOutputStreamOperator<Tuple2<String, Integer>> sum = stream.flatMap((String value, Collector<Tuple2<String,Integer>> out) -> {//value就是每一个元素的数据System.out.println("读取内容：" + value);//将每一个元素按照空格切分String[] split = value.split(" ");//遍历每一个单词for (String word : split) {//将每一个单词发送到下游out.collect(new Tuple2<>(word, 1));}}).returns(Types.TUPLE(Types.STRING, Types.INT)).keyBy(f->f.f0).sum(1);//打印结果sum.print();env.execute();}
}运行看结果
提交到集群运行
启动集群后我们使用命令
flink run -c 类全限定路径名 jar文件
flink run -c cn.wy.chapter02.WordCount FlinkDemo-1.0-SNAPSHOT.jar
可以看到任务提交切运行成功了
我们进入web-ui界面
网址
http://localhost:8081/#/job/completed
界面如下图所示
可以清晰的看到任务状态是FINISHED（完成）
任务执行成功了，我们的日志在哪看呢？
我们直接去TaskManager中看
点击地址进去
这里就是结果啦
【都看到这了，点点赞点点关注呗，爱你们】😚😚
💬
✨ 正在努力的小新~
💖 超级爱分享，分享各种有趣干货！
👩💻 提供：模拟面试 | 简历诊断 | 独家简历模板
🌈 感谢关注，关注了你就是我的超级粉丝啦！
🔒 以下内容仅对你可见~
作者：后端小知识，CSDN后端领域新星创作者 | 阿里云专家博主
CSDN个人主页：后端小知识
🔎GZH：后端小知识
🎉欢迎关注🔎点赞👍收藏⭐️留言📝