🍅程序员小王的博客：程序员小王的博客🍅欢迎点赞👍收藏⭐留言📝🍅如有编辑错误联系作者，如果有比较好的文章欢迎分享给我，我会取其精华去其糟粕前言这篇博客主要就是总结了各种flink部署方法及部署中出现的一些问题首先我进行了flink单机部署，个人建议不管是学习还是开发尽量不使用然后开始了flink自带集群部署，部署在三台服务器上，资源管理由flink集群自己管理，然后为了解决集群的单点故障问题，使用zookeeper监听事件，实现独立高可用集群，防止集群的单点故障，推荐这种集群可以部署在开发环境中测试使用最后一种就是flinkonyarn:把资源管理交给yarn实现，计算机资源统一由HaoopYA

🍅程序员小王的博客：程序员小王的博客🍅欢迎点赞👍收藏⭐留言📝🍅如有编辑错误联系作者，如果有比较好的文章欢迎分享给我，我会取其精华去其糟粕前言这篇博客主要就是总结了各种flink部署方法及部署中出现的一些问题首先我进行了flink单机部署，个人建议不管是学习还是开发尽量不使用然后开始了flink自带集群部署，部署在三台服务器上，资源管理由flink集群自己管理，然后为了解决集群的单点故障问题，使用zookeeper监听事件，实现独立高可用集群，防止集群的单点故障，推荐这种集群可以部署在开发环境中测试使用最后一种就是flinkonyarn:把资源管理交给yarn实现，计算机资源统一由HaoopYA

前言flink安装部署有三种方式local:单机模式，尽量不使用standalone:flink自带集群，资源管理由flink集群管理，开发环境测试使用，不需要hadoop集群flinkonyarn:把资源管理交给yarn实现，计算机资源统一由HaoopYARN管理，生产环境测试，需要先启动hadoop集群。（这里分为可以继续细分三种方式1.sessionmode长久启动一个flink集群接收job,main方法在客户端执行2.per-jobmode每个任务启动一个flink集群,main方法在客户端执行3.applicationmodeApplication模式为每个提交的应用程序创建一个集

一、版本doris：doris-1.2.3-rc02flink：flink1.4.6dinky：0.7.2jdk：1.8.0_191mysql：5.7二、安装doris官网下载地址：https://archive.apache.org/dist/doris/1.2/1.2.3-rc02/ #doris单机部署#创建doris目录mkdir/opt/module/doristarzxvfapache-doris-fe-1.2.3-bin-x86_64.tar.xz-C/opt/module/doristarzxvfapache-doris-be-1.2.3-bin-x86_64.tar.xz-C

今天这篇blog主要记录使用flink-sql对kafka中的数据进行过滤。以前对kafka数据进行实时处理时都是使用java来进行flink开发，需要创建一个工程，并且打成jar包再提交，流程固定但对于简单任务来说还是比较繁琐的。今天我们要对logstash采集到kafka中的数据进行过滤筛选，将筛选后的数据发送给另外一个kafkatopic，由于处理逻辑比较简单，使用flink自带的sql函数就可以搞定，所以我们今天就用flink-sql来解决这问题。问题描述我们需要筛选出ServiceA、ServiceB、ServiceC、ServiceD四个类打印出来的日志信息，并将目标信息发送到另外

flink实时计算任务可以从以下四个方面进行优化内存优化：Flink任务需要大量的内存来存储数据和状态信息。因此，我们需要尽可能地减少内存的使用量。可以通过以下几种方式来实现：使用更小的窗口大小：窗口大小越大，需要使用的内存就越多。因此，我们可以使用更小的窗口大小来减少内存的使用量。使用更小的数据类型：Flink支持多种数据类型，包括整数、浮点数、字符串等。如果我们只需要使用部分数据类型，那么就可以使用更小的数据类型来减少内存的使用量。使用更小的并行度：并行度越大，需要使用的内存就越多。因此，我们可以使用更小的并行度来减少内存的使用量。CPU优化：Flink任务需要大量的CPU资源来执行计算操

​当创建一个Flink任务后，该任务可能会经历多种状态。目前Flink给任务共定义了9种状态，包括：Created，Running，Finished，Cancelling，Canceled，Restarting，Failing，Failed，Suspended。下面这张图详细展示了一个Job可能会经历的所有状态。最简单的一种状态就是：作业启动时处于Created状态，任务启动后处于Running状态，当所有运行的task都进入了最终状态，任务会自动切换到Finished状态，这是一个正常job从启动到停止的正常流程，但再实际生产环境，可能也会出现其他一些情况。如果任务在启动后，因为网络原因或者

日常开发中都是用的简便json格式，但是偶尔也会遇到嵌套json的时候，因此在用flinksql的时候就有点麻烦，下面用简单例子简单定义处理下1，数据是网上摘抄，但包含里常用的大部分格式{   "afterColumns":{      "created":"1589186680",      "extra":{         "canGiving":false      },      "parameter":[1,2,3,4]   },   "beforeColumns":null,   "tableVersion":{      "binlogFile":null,      "bin

日常开发中都是用的简便json格式，但是偶尔也会遇到嵌套json的时候，因此在用flinksql的时候就有点麻烦，下面用简单例子简单定义处理下1，数据是网上摘抄，但包含里常用的大部分格式{   "afterColumns":{      "created":"1589186680",      "extra":{         "canGiving":false      },      "parameter":[1,2,3,4]   },   "beforeColumns":null,   "tableVersion":{      "binlogFile":null,      "bin

前言最近公司要搞搜索，需要把mysql中的数据同步到es中来进行搜索，由于公司已经搭建了flink集群，就打算用flink来做这个同步。本来以为很简单，跟着官网文档走就好了，结果没想到折腾了将近一周的时间……我也是没想到，这玩意网上资源竟然这么少，找到的全部都是通过flinksql-client实现的，但这有个问题，当fink集群重启，JOB就没有了，没有办法通过savePointing来恢复。所以还是记录下。代码直接上代码：publicstaticvoidmain(String[]args)throwsException{StreamExecutionEnvironmentenv=Strea