flink mysql集群增删改查

一、Flink 入门阶段最常见的疑问1.1、source/sink/mapFunction 处理的区别kafka作为source流数据处理需要查mysql查redis合并数据再查再处理再输出。对比表格Source/Sink 内查询 vs 独立 Map维度在 Source/Sink 内查询独立 RichMapFunction职责Source/Sink 承担了不属于它的逻辑每个算子职责单一并行度受限于 Source/Sink 的并行度Source 受分区数限制 原因参考1.1.2可独立设置并行度充分利用资源代码复用查询逻辑无法在其他作业复用增强函数可以独立复用测试需要模拟 Source/Sink 环境可以单独单元测试 Map 函数资源管理连接池生命周期与 Source/Sink 绑定容易错连接池在静态单例中Map 只引用无需管理关闭性能无法异步除非自己实现原因参考1.1.3可以轻松切换为异步AsyncDataStream可读性逻辑耦合维护困难清晰的数据流管道Source → Map → Sink一个 Flink 作业应该像一条流水线Source 进料 → 多个 Transformation增强、过滤、聚合→ Sink 出货。任何需要访问外部存储的逻辑都应该放在独立的 Transformation 算子中如RichMapFunction而不是塞进 Source 或 Sink。这样做的好处并行度自由调整资源生命周期可控代码易于理解和维护符合 Flink 的最佳实践1.1.2、为什么 Source 和 Sink 会受“分区”限制1.1.2.1 Source 的并行度 ≤ Kafka 分区数Kafka 一个分区只能被同组一个消费者消费Flink Kafka Source 并行度 消费者实例数。若并行度 Kafka 分区数多余并行实例会空闲无法提升效率。结论Source 并行度上限由 Kafka 分区数决定不能随便调大。1.1.2.2 Sink 的并行度不一定受分区限制但有其他约束写入数据库 Sink如 JDBC时并行度受数据库连接池大小、目标表锁等影响不是受分区限制而是受资源限制。1.1.3、source/sink不支持异步Flink 里 Source 和 Sink不是不能异步而是不推荐这么做会破坏exactly-once、checkpoint、事务等一致性语义实现复杂易出错Source/Sink 本身是按序拉取 / 写入不需要中间算子那种多元素并发模式强行在 Source 里查库会阻塞吞吐低而AsyncDataStream 才是专门给中间算子做异步查询、提升吞吐的标准方案。1.1.4、逻辑耦合在 Source/Sink 的危害把逻辑写在 Source 里会导致复用性差逻辑和 Kafka 绑定其他数据源无法复用测试麻烦必须启动 Kafka、MySQL 才能测维护风险大改逻辑要动 Source易影响数据读取拆成独立 Map 算子逻辑通用可复用单元测试简单修改安全、风险小维度写在 Source/Sink写在独立 Map 算子并行度灵活性Source 受 Kafka 分区数限制可独立调大例如设为 100异步 I/O 支持极难实现不推荐直接使用AsyncDataStream作业间复用无法复用任何数据源都能使用该 Map 函数测试难度需要外部系统Kafka、MySQL纯单元测试即可职责清晰度Source 既读数据又做业务混乱一个算子只做一件事容错恢复Checkpoint 复杂无状态恢复简单1.1.5、正确模块使用指南Source只做Kafka → KafkaParamVOMap 算子做KafkaParamVO → (查 MySQL 两次 Redis) → EnrichedResult如果是同步查询用RichMapFunction如果是高吞吐异步用AsyncDataStream 自定义RichAsyncFunctionSink只做EnrichedResult → 目标系统这样你的作业既稳定又能发挥 Flink 的并行优势还能随意复用增强逻辑。二、mysql查询实践三、flink集群日志管理你只需要在Kubernetes里完成下面几步准备配置文件根据你使用的日志框架如 Log4j2创建一个包含滚动策略的配置文件例如log4j.properties。可以参照前文方案一中的配置示例。创建 ConfigMap在K8s集群中创建一个 ConfigMap 来存放你的日志配置文件。挂载 ConfigMap在 JobManager 和 TaskManager 的 Pod 定义中将这个 ConfigMap 以文件形式挂载覆盖 Flink 容器内/opt/flink/conf/log4j.properties的默认配置。滚动更新更新你的 Flink 集群或任务使配置生效。四、flink负载均衡测试五、如何丝滑升级flink任务在K8s上管理Flink任务要实现“丝滑”的提交与升级最推荐的方式是采用Flink Kubernetes Operator。它通过K8s的声明式APIYAML文件让你像管理其他K8s应用一样管理Flink任务从而实现自动化、声明式、无中断的任务升级。 如何提交任务在K8s上提交Flink任务主要有三种方式以下是它们的对比和选择建议方式简介优点缺点适用场景1. Flink Kubernetes Operator (强烈推荐)通过声明式YAML文件定义任务Operator自动完成资源创建和生命周期管理。声明式管理契合K8s理念支持自动化、无中断升级简化运维生产首选。需要额外安装和了解Operator机制。生产环境追求高自动化和稳定性的团队。2. Flink Native Client使用Flink自带的命令行工具直接提交任务到K8s集群。直接无需安装Operator快速上手适合初期测试。升级、停止等操作需手动管理不够“K8s原生”脚本方式难以管理大量任务。开发测试或初期小规模使用。3. Flink Web UI / REST API通过Web界面上传JAR包提交任务。图形化操作直观调试方便。不适合自动化运维不适合生产环境。临时调试或非生产环境。 如何实现“丝滑”升级任务升级的核心是保证状态不丢失和服务不中断关键在于Savepoint。它就像是给运行中的任务照了一张“快照”保存了当前所有的计算状态升级时新任务可以直接从这个“快照”中恢复无缝衔接工作。Operator通过upgradeMode参数控制升级时的行为主要有三种模式升级模式 (upgradeMode)运作方式生产环境推荐度savepoint最安全的方式。升级时自动创建一个Savepoint新任务从该Savepoint恢复。如果任务运行健康能保证状态100%不丢失。✅强烈推荐last-state快速升级。直接从最后一个Checkpoint恢复无需额外创建Savepoint。即使任务异常也可用。✅推荐stateless无状态升级。不保存任何状态任务从零开始。⚠️ 仅限无状态任务配置示例与操作步骤假设你有一个flink-deployment.yaml文件用于定义一个Application模式的Flink任务。配置Savepoint存储首先你需要为Savepoint和Checkpoint配置一个持久化存储如S3, HDFS, NFS等并在flinkConfiguration中指定。yamlspec: flinkConfiguration: state.savepoints.dir: s3://your-bucket/flink-savepoints state.checkpoints.dir: s3://your-bucket/flink-checkpoints high-availability.storageDir: s3://your-bucket/flink-ha定义升级模式在spec.job中设置upgradeMode: savepoint。当需要升级任务时你只需要更新YAML文件中的镜像、JAR包或并行度等配置然后重新apply即可。yamlspec: job: upgradeMode: savepoint jarURI: local:///opt/flink/usrlib/your-new-flink-job.jar # 指向新版本的JAR parallelism: 2 # 假设你将并行度从1改为了2 state: running执行升级运行以下命令即可触发自动化升级流程。bashkubectl apply -f flink-deployment.yaml执行后Operator会自动完成“创建Savepoint - 停止旧任务 - 从Savepoint启动新任务”的全流程实现无缝升级。 核心注意事项高可用HA是基石在生产环境中务必为Flink集群和Operator本身配置HA避免单点故障。状态兼容性需谨慎升级任务时如果修改了状态的Schema新任务可能无法读取旧的Savepoint。升级前请务必验证状态兼容性。从零到一的规划搭建存储准备好S3、MinIO等持久化存储。安装Operator通过Helm或YAML在K8s集群中安装Flink Kubernetes Operator。部署任务编写YAML文件配置好HA和upgradeMode使用kubectl apply -f提交任务。升级任务修改YAML文件中的业务配置如镜像、并行度再次执行kubectl apply -f。总的来说生产环境的最佳实践是使用Flink Kubernetes Operator Application模式 upgradeMode: savepoint 远程持久化存储。这套方案能很好地帮你实现自动化、高可用的任务管理。