Flink从入门到实践(一):Flink入门、Flink部署

文章目录

  • 系列文章索引
  • 一、快速上手
    • 1、导包
    • 2、求词频demo
      • (1)要读取的数据
      • (2)demo1:批处理(离线处理)
      • (3)demo2 - lambda优化:批处理(离线处理)
      • (4)demo3:流处理(实时处理)
      • (5)总结:实时vs离线
      • (6)demo4:批流一体
      • (7)对接Socket
  • 二、Flink部署
    • 1、Flink架构
    • 2、Standalone部署
    • 3、自运行flink-web
    • 4、通过参数传递
    • 5、通过webui提交job
    • 6、停止作业
    • 7、常用命令
    • 8、集群
  • 参考资料

系列文章索引

Flink从入门到实践(一):Flink入门、Flink部署
Flink从入门到实践(二):Flink DataStream API
Flink从入门到实践(三):数据实时采集 - Flink MySQL CDC

一、快速上手

1、导包

<!-- fink 相关依赖 -->
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><version>1.18.0</version>
</dependency>

2、求词频demo

注意!自Flink 1.18以来,所有Flink DataSet api都已弃用,并将在未来的Flink主版本中删除。您仍然可以在DataSet中构建应用程序,但是您应该转向DataStream和/或Table API。

(1)要读取的数据

定义data内容:

pk,pk,pk
ruoze,ruoze
hello

(2)demo1:批处理(离线处理)

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;/*** 使用Flink进行批处理,并统计wc*** 结果:* (bye,2)* (hello,3)* (hi,1)*/
public class BatchWordCountApp {public static void main(String[] args) throws Exception {// step0: Spark中有上下文,Flink中也有上下文,MR中也有ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();// step1: 读取文件内容  ==> 一行一行的字符串而已DataSource<String> source = env.readTextFile("data/wc.data");// step2: 每一行的内容按照指定的分隔符进行拆分  1:Nsource.flatMap(new FlatMapFunction<String, String>() {/**** @param value 读取到的每一行数据* @param out 输出的集合*/@Overridepublic void flatMap(String value, Collector<String> out) throws Exception {// 使用,进行分割String[] splits = value.split(",");for(String split : splits) {out.collect(split.toLowerCase().trim());}}}).map(new MapFunction<String, Tuple2<String,Integer>>() {/**** @param value 每一个元素 (hello, 1)(hello, 1)(hello, 1)*/@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {return Tuple2.of(value, 1);}}).groupBy(0)  // step4: 按照单词进行分组  groupBy是离线的api,传下标.sum(1)  // ==> 求词频 sum,传下标.print(); // 打印}
}

(3)demo2 - lambda优化:批处理(离线处理)

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;/*** lambda表达式优化*/
public class BatchWordCountAppV2 {public static void main(String[] args) throws Exception {ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();DataSource<String> source = env.readTextFile("data/wc.data");/*** lambda语法: (参数1,参数2,参数3...) -> {函数体}*/
//        source.map(String::toUpperCase).print();// 使用了Java泛型,由于泛型擦除的原因,需要显示的声明类型信息source.flatMap((String value, Collector<Tuple2<String,Integer>> out) -> {String[] splits = value.split(",");for(String split : splits) {out.collect(Tuple2.of(split.trim(), 1));}}).returns(Types.TUPLE(Types.STRING, Types.INT)).groupBy(0).sum(1).print();}
}

(4)demo3:流处理(实时处理)

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;/*** 流式处理* 结果:* 8> (hi,1)* 6> (hello,1)* 5> (bye,1)* 6> (hello,2)* 6> (hello,3)* 5> (bye,2)*/
public class StreamWCApp {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> source = env.readTextFile("data/wc.data");source.flatMap((String value, Collector<Tuple2<String,Integer>> out) -> {String[] splits = value.split(",");for(String split : splits) {out.collect(Tuple2.of(split.trim(), 1));}}).returns(Types.TUPLE(Types.STRING, Types.INT)).keyBy(x -> x.f0) // 这种写法一定要掌握!流式的并没有groupBy,而是keyBy!根据第一个值进行sum.sum(1).print();// 需要手动开启env.execute("作业名字");}
}

(5)总结:实时vs离线

离线:结果是一次性出来的。
实时:来一个数据处理一次,数据是带状态的。

(6)demo4:批流一体

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;/*** 采用批流一体的方式进行处理*/
public class FlinkWordCountApp {public static void main(String[] args) throws Exception {// 统一使用StreamExecutionEnvironment这个执行上下文环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC); // 选择处理方式 批/流/自动DataStreamSource<String> source = env.readTextFile("data/wc.data");source.flatMap((String value, Collector<Tuple2<String,Integer>> out) -> {String[] splits = value.split(",");for(String split : splits) {out.collect(Tuple2.of(split.trim(), 1));}}).returns(Types.TUPLE(Types.STRING, Types.INT)).keyBy(x -> x.f0) // 这种写法一定要掌握.sum(1).print();// 执行env.execute("作业名字");}
}

(7)对接Socket

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;/*** 使用Flink对接Socket的数据并进行词频统计** 大数据处理的三段论: 输入  处理  输出**/
public class FlinkSocket {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();/*** 数据源:可以通过多种不同的数据源接入数据:socket  kafka  text** 官网上描述的是 env.addSource(...)** socket的方式对应的并行度是1,因为它来自于SourceFunction的实现*/DataStreamSource<String> source = env.socketTextStream("localhost", 9527);System.out.println(source.getParallelism());// 处理source.flatMap((String value, Collector<Tuple2<String,Integer>> out) -> {String[] splits = value.split(",");for(String split : splits) {out.collect(Tuple2.of(split.trim(), 1));}}).returns(Types.TUPLE(Types.STRING, Types.INT)).keyBy(x -> x.f0) // 这种写法一定要掌握.sum(1)// 数据输出.print();  // 输出到外部系统中去env.execute("作业名字");}
}

二、Flink部署

1、Flink架构

https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/concepts/flink-architecture/
Flink是一个分布式的带有状态管理的计算框架,可以运行在常用/常见的集群资源管理器上(YARN、K8S)。

一个JobManager(协调/分配),一个或多个TaskManager(工作)。
在这里插入图片描述
在这里插入图片描述

2、Standalone部署

按照官网下载执行即可:
https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/try-flink/local_installation/

可以根据官网来安装,需要下载、解压、安装。
也可以使用docker安装。

启动之后,localhost:8081就可以访问管控台了。

3、自运行flink-web

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-runtime-web</artifactId><version>1.18.0</version>
</dependency>
Configuration configuration = new Configuration();
configuration.setInteger("rest.port", 8082); // 指定web端口,开启webUI,不写的话默认8081
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
// 新版本可以直接使用getExecutionEnvironment(conf)

以上亲测并不好使……具体原因未知,设置为flink1.16版本或许就好用了。

4、通过参数传递

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 通过参数传递进来Flink引用程序所需要的参数,flink自带的工具类
ParameterTool tool = ParameterTool.fromArgs(args);
String host = tool.get("host");
int port = tool.getInt("port");DataStreamSource<String> source = env.socketTextStream(host, port);
System.out.println(source.getParallelism());

可以通过命令行参数:–host localhost --port 8765

5、通过webui提交job

在这里插入图片描述
在这里插入图片描述

6、停止作业

在这里插入图片描述

7、常用命令

# 查看作业列表
flink list -a  # 所有
flink list -r  # 正在运行的
# 停止作业
flink cancel <jobid># 提交job
# -c,--class <classname> 指定main方法
# -C,--classpath <url> 指定classpath
# -p,--parallelism <paralle> 指定并行度
flink run -c com.demo.FlinkDemo FlinkTest.jar 

8、集群

https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/concepts/flink-architecture/#flink-application-execution

https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/deployment/overview/

单机部署Session Mode和Application Mode:
https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/deployment/resource-providers/standalone/overview/

k8s:
https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/deployment/resource-providers/native_kubernetes/

YARN:
https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/deployment/resource-providers/yarn/

参考资料

https://flink.apache.org/
https://nightlies.apache.org/flink/flink-docs-stable/

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/254751.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【机器学习】数据清洗之识别缺失点

&#x1f388;个人主页&#xff1a;甜美的江 &#x1f389;欢迎 &#x1f44d;点赞✍评论⭐收藏 &#x1f917;收录专栏&#xff1a;机器学习 &#x1f91d;希望本文对您有所裨益&#xff0c;如有不足之处&#xff0c;欢迎在评论区提出指正&#xff0c;让我们共同学习、交流进步…

[神奇代码岛】皮肤功能使用

前言 最近有很多人在制作地图的时候&#xff0c;因该会用到皮肤的功能&#xff0c;但是皮肤操作只知道UI操作&#xff0c;但缺点是&#xff0c;只能设置地图默认皮肤&#xff0c;根本都做不到想要的什么皮肤购买功能&#xff0c;自主穿戴功能&#xff0c;而API官方又放在非常隐…

老胡的周刊(第128期)

老胡的信息周刊[1]&#xff0c;记录这周我看到的有价值的信息&#xff0c;主要针对计算机领域&#xff0c;内容主题极大程度被我个人喜好主导。这个项目核心目的在于记录让自己有印象的信息做一个留存以及共享。 &#x1f3af; 项目 coze-discord-proxy[2] 代理 Discord-Bot 对…

微服务OAuth 2.1扩展额外信息到JWT并解析(Spring Security 6)

文章目录 一、简介二、重写UserDetailsService三、Controller解析JWT获取用户信息四、后记 一、简介 VersionJava17SpringCloud2023.0.0SpringBoot3.2.1Spring Authorization Server1.2.1Spring Security6.2.1mysql8.2.0 Spring Authorization Server 使用JWT时&#xff0c;前…

PbootCMS采集插件使用教程

这篇Pboot采集教程教你使用PbootCMS采集插件&#xff0c;自动批量采集网页文章数据&#xff0c;并发布到PbootCMS系统&#xff0c;快速丰富网站的内容。 1. 下载并安装PbootCMS采集插件 1-1&#xff09;PbootCMS采集插件免费下载&#xff1a;Pboot采集插件-PbootCMS发布模块下…

【Docker】私有仓库

目录 1.搭建 2. 上传镜像 3.拉取镜像 1.搭建 1.拉取私有仓库的镜像 docker pull registry 2.创建私有仓库容器 docker run -id --nameregistry -p 5000:5000 registry 3.打开浏览器,输入地址&#xff08;http:私有仓库服务器ip:5000/v2/_catalog&#xff09; 出现如图表示私…

如何运行心理学知识(心流)来指导工作和生活

如何运用心流来指导工作和生活 如何联系我 作者&#xff1a;鲁伟林 邮箱&#xff1a;thinking_fioa163.com或vlinyes163.com GitHub&#xff1a;https://github.com/thinkingfioa/ReadingSummary 版权声明&#xff1a;文章和记录为个人所有&#xff0c;如果转载或个人学习…

TDengine用户权限管理

Background 官方文档关于用户管理没有很详细的介绍&#xff0c;只有零碎的几条&#xff0c;这里记录下方便后面使用。官方文档&#xff1a;https://docs.taosdata.com/taos-sql/show/#show-users 1、查看用户 show users;super 1&#xff0c;表示超级用户权限 0&#xff0c;表…

MySQL用心总结

大家好&#xff0c;好久不见&#xff0c;今天笔者用心一步步写一份mysql的基础操作指南&#xff0c;欢迎各位点赞收藏 -- 启动MySQL net start mysql-- 创建Windows服务 sc create mysql binPath mysqld_bin_path(注意&#xff1a;等号与值之间有空格) mysql -h 地址 -…

kmeans聚类选择最优K值python实现

Kmeans算法中K值的确定是很重要的。 下面利用python中sklearn模块进行数据聚类的K值选择 数据集自制数据集&#xff0c;格式如下&#xff1a; 维度为3。 ①手肘法 手肘法的核心指标是SSE(sum of the squared errors&#xff0c;误差平方和)&#xff0c; 其中&#xff0c;Ci是第…

【数据分享】1929-2023年全球站点的逐年平均降水量(Shp\Excel\免费获取)

气象数据是在各项研究中都经常使用的数据&#xff0c;气象指标包括气温、风速、降水、湿度等指标&#xff0c;说到常用的降水数据&#xff0c;最详细的降水数据是具体到气象监测站点的降水数据&#xff01; 有关气象指标的监测站点数据&#xff0c;之前我们分享过1929-2023年全…

华为视频监控接入到视频监控平台 (华为网路监控摄像机IPC和华为视频节点设备VCN)

目 录 一、设备介绍 1.1 华为VCN介绍 1.2 AS-V1000视频监控平台介绍 1.3 平台服务器配置说明 二、安装、配置HW_IVS软件 2.1下载安装HW_IVS软件 2.2登录HW_IVS 2.3共享到外域 三、配置华为外域参数 3.1 PCG模块设置 3.2通信协议GBT28181配置 3.3传…

k8s学习(RKE+k8s+rancher2.x)成长系列之简配版环境搭建(二)

三、简配版集群&#xff0c;适用于demo环境 1.集群架构设计 主机名角色配置(核数&#xff0c;内存&#xff0c;磁盘)MasterRKE,controlplane,etcd,worker,rancher-master2C 8G 40GSlaver1controlplane,worker,rancher-master2C 8G 40GSlaver2controlplane,worker,rancher-mas…

[office] 怎么在Excel2003菜单栏自定义一个选项卡 #其他#微信#知识分享

怎么在Excel2003菜单栏自定义一个选项卡 怎么在Excel2003菜单栏自定义一个选项卡 ①启动Excel2003&#xff0c;单击菜单栏--工具--自定义。 ②在自定义界面&#xff0c;我们单击命令标签&#xff0c;在类别中选择新菜单&#xff0c;鼠标左键按住新菜单&#xff0c;拖放到菜单栏…

MATLAB环境下一维时间序列信号的同步压缩小波包变换

时频分析相较于目前的时域、频域信号处理方法在分析时变信号方面&#xff0c;其主要优势在于可以同时提供时域和频域等多域信号信息&#xff0c;并清晰的刻画了频率随时间的变化规律&#xff0c;已被广泛用于医学工程、地震、雷达、生物及机械等领域。 线性时频分析方法是将信…

tee漏洞学习-翻译-2:探索 Qualcomm TrustZone的实现

原文&#xff1a;http://bits-please.blogspot.com/2015/08/exploring-qualcomms-trustzone.html 获取 TrustZone image 从两个不同的位置提取image 从手机设备本身从google factory image 已经root的Nexus 5设备&#xff0c;image存储在eMMC芯片上&#xff0c;并且eMMC芯片…

探索Spring Validation:优雅实现后端数据验证的艺术

在现代Web应用开发中&#xff0c;数据验证是一项至关重要的任务&#xff0c;确保应用程序接收到的用户输入符合预期规范&#xff0c;不仅能够提高系统的健壮性&#xff0c;也能有效防止潜在的安全漏洞。Spring Framework通过其内置的Spring Validation模块&#xff0c;为我们提…

极值图论基础

目录 一&#xff0c;普通子图禁图 二&#xff0c;Turan问题 三&#xff0c;Turan定理、Turan图 1&#xff0c;Turan定理 2&#xff0c;Turan图 四&#xff0c;以完全二部图为禁图的Turan问题 1&#xff0c;最大边数的上界 2&#xff0c;最大边数的下界 五&#xff0c;…

Vue3大事件项目(ing)

文章目录 核心内容1.大事件项目介绍2.大事件项目创建3.Eslint配置代码风格4.配置代码检查工作流问题: pnpm lint是全量检查,耗时问题,历史问题 5.目录调整6.vue-router4 路由代码解析7.引入 Element Plus 组件库8.Pinia 构建仓库 和 持久化9.Pinia 仓库统一管理 核心内容 Vue3…

【C/C++ 10】扫雷小游戏

一、题目 写一个扫雷小游戏&#xff0c;每次输入一个坐标&#xff0c;若该处是地雷&#xff0c;则游戏失败&#xff0c;若该处不是地雷&#xff0c;则显示周围地雷数量&#xff0c;若扫除全部非地雷区域&#xff0c;则扫雷成功。 二、算法 设置两张地图&#xff08;二维数组&…