当前位置: 首页 > news >正文

Flink 中 JDBC Connector 使用详解

1. 背景

在实时计算或离线任务中,往往需要与关系型数据库交互,例如 MySQL、PostgreSQL 等。Apache Flink 提供了 JDBC Connector,可以方便地将流式数据写入或读取数据库。

本文将介绍 Flink JDBC Connector 的基础用法、配置方法以及注意事项,帮助开发者更好地集成数据库操作。


2. JDBC Connector 的基础概念

JDBC Connector 是 Flink 官方提供的一个用于连接关系型数据库的工具包,支持:

  • Source:从数据库读取数据。
  • Sink:将数据写入数据库。

使用 JDBC Connector 可以实现对数据库的实时写入,也可以用作批量操作的工具。


3. Maven 依赖

在项目中添加 Flink JDBC 依赖:

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc_2.12</artifactId><version>1.17.0</version> <!-- 根据实际使用的 Flink 版本调整 -->
</dependency>

如果使用 MySQL 数据库,还需添加 MySQL 驱动:

<dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.33</version> <!-- MySQL 驱动版本 -->
</dependency>

4. JDBC Connector 的使用

4.1 写入数据库(Sink)

以下是一个将流式数据写入 MySQL 的示例:

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.connector.jdbc.JdbcSink;public class JdbcSinkExample {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 模拟输入数据env.fromElements(Tuple2.of(1, "Alice"),Tuple2.of(2, "Bob"),Tuple2.of(3, "Charlie")).addSink(JdbcSink.sink("INSERT INTO users (id, name) VALUES (?, ?)", // SQL 语句(ps, t) -> {ps.setInt(1, t.f0);  // 设置第一个参数为 IDps.setString(2, t.f1);  // 设置第二个参数为 Name},JdbcSink.DefaultJdbcExecutionOptions.builder().withBatchSize(100) // 批量写入大小.build(),() -> JdbcSink.defaultJdbcConnectionProvider("jdbc:mysql://localhost:3306/testdb", // 数据库 URL"root",  // 用户名"password" // 密码)));env.execute("Flink JDBC Sink Example");}
}
关键点解析
  1. SQL 语句:支持动态参数 ? 占位符,适合批量插入。
  2. 参数绑定:通过 Lambda 表达式绑定输入数据与 SQL 参数。
  3. 批量写入:通过 JdbcExecutionOptions 配置批量写入策略。

4.2 从数据库读取数据(Source)

以下是一个从 MySQL 读取数据并打印的示例:

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;public class JdbcSourceExample {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<Tuple2<Integer, String>> sourceStream = env.createInput(JdbcInputFormat.buildJdbcInputFormat().setDrivername("com.mysql.cj.jdbc.Driver") // JDBC 驱动.setDBUrl("jdbc:mysql://localhost:3306/testdb") // 数据库 URL.setUsername("root") // 用户名.setPassword("password") // 密码.setQuery("SELECT id, name FROM users") // SQL 查询.setRowTypeInfo(Types.TUPLE(Types.INT, Types.STRING)) // 结果类型.finish());sourceStream.print();env.execute("Flink JDBC Source Example");}
}
关键点解析
  1. SQL 查询:需要提供完整的查询语句。
  2. 结果类型:通过 RowTypeInfo 显式定义数据库返回的数据结构。

5. JDBC Connector 的配置选项

5.1 批量写入配置

通过 JdbcExecutionOptions 可调整写入策略:

  • withBatchSize(int):设置批量写入大小(默认为 500)。
  • withBatchIntervalMs(long):设置批量写入的时间间隔。
  • withMaxRetries(int):设置写入失败后的最大重试次数。

5.2 数据库连接池

Flink JDBC Connector 默认使用单个连接执行操作。对于高并发需求,可以结合 HikariCP 等连接池框架优化性能。


6. 注意事项

  1. 事务支持

    • 默认情况下,JDBC Sink 使用批量提交,未显式开启事务。如果需要事务一致性,可以通过 JDBC 驱动自行管理事务。
  2. 数据库性能瓶颈

    • 数据库可能成为瓶颈,建议使用批量写入和合适的索引优化性能。
    • 高写入场景可考虑切换到 Kafka、HBase 等专为实时写入设计的存储系统。
  3. 错误处理

    • 可通过 withMaxRetries 设置重试次数。
    • 对于未能成功写入的数据,可考虑使用侧输出流保存以供后续处理。
  4. 分布式读取

    • 默认情况下,Flink JDBC Source 在单线程上运行,性能可能有限。可以使用分片或其他工具提升读取性能。

7. 总结

Flink JDBC Connector 是一个简单而高效的工具,适用于实时计算场景下与关系型数据库的交互。无论是数据写入还是读取,都可以通过简单配置快速实现。但对于高并发和大规模数据场景,需要根据业务需求调整策略。

相关文章:

Flink 中 JDBC Connector 使用详解

1. 背景 在实时计算或离线任务中&#xff0c;往往需要与关系型数据库交互&#xff0c;例如 MySQL、PostgreSQL 等。Apache Flink 提供了 JDBC Connector&#xff0c;可以方便地将流式数据写入或读取数据库。 本文将介绍 Flink JDBC Connector 的基础用法、配置方法以及注意事…...

【Linux打怪升级记 | 报错02】-bash: 警告:setlocale: LC_TIME: 无法改变区域选项 (zh_CN.UTF-8)

&#x1f5fa;️博客地图 &#x1f4cd;1、报错发现 &#x1f4cd;2、原因分析 &#x1f4cd;3、解决办法 &#x1f4cd;4、测试结果 1、报错发现 装好了CentOS操作系统&#xff0c;使用ssh远程登陆CentOS&#xff0c;出现如下告警信息&#xff1a; bash: 警告:setlocale…...

未来已来?AI技术革新改变我们的生活

在21世纪的今天&#xff0c;人工智能&#xff08;AI&#xff09;不再是一个遥远的概念&#xff0c;而是逐渐渗透到我们生活的方方面面。从智能家居到自动驾驶汽车&#xff0c;从个性化推荐系统到医疗诊断辅助&#xff0c;AI技术正在以惊人的速度发展&#xff0c;并深刻地影响着…...

【Linux】进程的生命之旅——诞生、消逝与守候(fork/exit/wait)

&#x1f3ac; 个人主页&#xff1a;谁在夜里看海. &#x1f4d6; 个人专栏&#xff1a;《C系列》《Linux系列》《算法系列》 ⛰️ 一念既出&#xff0c;万山无阻 目录 &#x1f4d6;一、进程创建 1.fork函数 &#x1f4da;高层封装特性 &#x1f4da;fork返回值 2.写时拷…...

使用vcpkg自动链接tinyxml2时莫名链接其他库(例如boost)

使用vcpkg自动链接tinyxml2时莫名链接其他库&#xff08;例如boost&#xff09; vcpkg的自动链接功能非常方便&#xff0c;但在某些情况下会出现过度链接的问题。 链接错误症状 以tinyxml2为例&#xff0c;程序中调用tinyxml2的函数后&#xff0c;若vcpkg中同时存在opencv和…...

【去毛刺】OpenCV图像处理基础:腐蚀与膨胀操作入门

在数字图像处理中&#xff0c;形态学操作是一种常用的技术&#xff0c;用于提取图像中的特定形状或特征。其中&#xff0c;腐蚀&#xff08;Erosion&#xff09;和膨胀&#xff08;Dilation&#xff09;是两种基本的形态学运算。本文将通过一个简单的例子来演示如何使用Python中…...

道可云人工智能元宇宙每日资讯|第三届京西地区发展论坛成功召开

道可云元宇宙每日简报&#xff08;2024年11月27日&#xff09;讯&#xff0c;今日元宇宙新鲜事有&#xff1a; 工信部等十二部门印发《5G规模化应用“扬帆”行动升级方案》 11月25日&#xff0c;工业和信息化部等十二部门印发《5G规模化应用“扬帆”行动升级方案》。《方案》…...

若依框架部署在网站一个子目录下(/admin)问题(

部署在子目录下首先修改vue.config.js文件&#xff1a; 问题一&#xff1a;登陆之后跳转到了404页面问题&#xff0c;解决办法如下&#xff1a; src/router/index.js 把404页面直接变成了首页&#xff08;大佬有啥优雅的解决办法求告知&#xff09; 问题二&#xff1a;退出登录…...

【ue5】UE5运行时下载视频/UE5 runtime download video(MP4)

插件还是老朋友。 节点的content type要打对。 &#xff08;参照表&#xff1a;MIME 类型&#xff08;MIME Type&#xff09;完整对照表 - 免费在线工具&#xff09; 结果展示&#xff1a;...

对比C++,Rust在内存安全上做的努力

简介 近年来&#xff0c;越来越多的组织表示&#xff0c;如果新项目在技术选型时需要使用系统级开发语言&#xff0c;那么不要选择使用C/C这种内存不安全的系统语言&#xff0c;推荐使用内存安全的Rust作为替代。 谷歌也声称&#xff0c;Android 的安全漏洞&#xff0c;从 20…...

如何利用 Qt 的模块化架构组织大型项目

目录 1. 大型项目的架构设计 1.1 分层架构 1.2 事件驱动与异步架构 2. 模块划分与职责分离 2.1 功能模块划分 2.2 模块之间的依赖管理 3. 跨平台开发与模块复用 在大型软件项目中&#xff0c;随着代码量的增加和功能的扩展&#xff0c;项目的复杂度会显著提升。没有良好…...

探索Python词云库WordCloud的奥秘

文章目录 探索Python词云库WordCloud的奥秘1. 背景介绍&#xff1a;为何选择WordCloud&#xff1f;2. WordCloud库简介3. 安装WordCloud库4. 简单函数使用方法5. 应用场景示例6. 常见Bug及解决方案7. 总结 探索Python词云库WordCloud的奥秘 1. 背景介绍&#xff1a;为何选择Wo…...

MySQL根据idb文件恢复数据

首先得有对应表的idb文件以及建表语句 1.首先在新数据库建表 CREATE TABLE sys_menu (id bigint(20) NOT NULL,parent_id bigint(20) NULL DEFAULT NULL,name varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL,type int(11) NULL DEFAULT …...

hadoop-mapreduce词频统计

一、Map Reduce主要阶段 二、词频统计示例 0.MapReduce 词频统计(Word Count)示例图 1. Input 阶段&#xff08;输入阶段&#xff09; 输入数据是一段文本&#xff0c;如下&#xff1a; Hadoop is a big data framework. Hadoop can store vast data. Hadoop processes big …...

精心修炼Java并发编程(JUC)-volatile与synchronized关键字

volatile volatile 是 JVM 提供的 最轻量级的同步机制&#xff0c;中文意思是不稳定的&#xff0c;易变的&#xff0c;用 volatile 修饰变量是为了保证变量在多线程中的可见性&#xff0c;它表达的含义是&#xff1a;告诉编译器&#xff0c;对这个变量的读写&#xff0c;需要基…...

【ROS2】ROS2 与 ROS1 编码方式对比(Python实现)

目录 一、初始化和关闭节点二、发布者三、订阅者四、服务端五、客户端六、参数管理七、日志记录八、生命周期管理 ROS2 在 Python 编程中引入了一些新的概念和 API&#xff0c;这些变化使得代码更加模块化和易于维护。特别是 rclpy 库提供了更丰富的功能和更好的错误处理机制&a…...

ElasticSearch的下载和基本使用(通过apifox)

1.概述 一个开源的高扩展的分布式全文检索引擎&#xff0c;近乎实时的存储&#xff0c;检索数据 2.安装路径 Elasticsearch 7.8.0 | Elastic 安装后启动elasticsearch-7.8.0\bin里的elasticsearch.bat文件&#xff0c; 启动后就可以访问本地的es库http://localhost:9200/ …...

城市轨道交通运营控制指挥中心设计方案

为某城市轨道交通运营控制指挥中心(OCC)的设计提供方案时,我们需要考虑到多个方面的需求,包括系统架构、设备选择、功能实现、数据流与监控、通信管理等。以下是一个综合性的设计方案,涉及系统硬件和软件的选择、布局规划、安全性等方面,以确保指挥中心的高效运作、实时监…...

多目标优化算法:多目标河马优化算法(MOHOA)求解ZDT1、ZDT2、ZDT3、ZDT4、ZDT6,提供完整MATLAB代码

一、河马优化算法 河马优化算法&#xff08;Hippopotamus optimization algorithm&#xff0c;HO&#xff09;由Amiri等人于2024年提出的一种模拟自然界中河马觅食行为的新型群体智能优化算法。该算法由Mohammad Hussein Amiri等人于2024年2月发表在Nature旗下子刊《Scientifi…...

线程与进程的个人理解

进程&#xff08;Process&#xff09;&#xff1a; 一个程序在执行时&#xff0c;操作系统为其分配的资源&#xff08;如内存、CPU 时间等&#xff09;构成了一个进程。每个进程都有自己的独立的地址空间、堆栈和局部变量&#xff0c;它们之间不共享内存&#xff08;除非通过特…...

【Linux】shell脚本忽略错误继续执行

在 shell 脚本中&#xff0c;可以使用 set -e 命令来设置脚本在遇到错误时退出执行。如果你希望脚本忽略错误并继续执行&#xff0c;可以在脚本开头添加 set e 命令来取消该设置。 举例1 #!/bin/bash# 取消 set -e 的设置 set e# 执行命令&#xff0c;并忽略错误 rm somefile…...

Cesium1.95中高性能加载1500个点

一、基本方式&#xff1a; 图标使用.png比.svg性能要好 <template><div id"cesiumContainer"></div><div class"toolbar"><button id"resetButton">重新生成点</button><span id"countDisplay&qu…...

前端导出带有合并单元格的列表

// 导出async function exportExcel(fileName "共识调整.xlsx") {// 所有数据const exportData await getAllMainData();// 表头内容let fitstTitleList [];const secondTitleList [];allColumns.value.forEach(column > {if (!column.children) {fitstTitleL…...

2024年赣州旅游投资集团社会招聘笔试真

2024年赣州旅游投资集团社会招聘笔试真 题 ( 满 分 1 0 0 分 时 间 1 2 0 分 钟 ) 一、单选题(每题只有一个正确答案,答错、不答或多答均不得分) 1.纪要的特点不包括()。 A.概括重点 B.指导传达 C. 客观纪实 D.有言必录 【答案】: D 2.1864年,()预言了电磁波的存在,并指出…...

macOS多出来了:Google云端硬盘、YouTube、表格、幻灯片、Gmail、Google文档等应用

文章目录 问题现象问题原因解决办法 问题现象 macOS启动台&#xff08;Launchpad&#xff09;多出来了&#xff1a;Google云端硬盘、YouTube、表格、幻灯片、Gmail、Google文档等应用。 问题原因 很明显&#xff0c;都是Google家的办公全家桶。这些应用并不是通过独立安装的…...

Mysql8 忘记密码重置,以及问题解决

1.使用免密登录 找到配置MySQL文件&#xff0c;我的文件路径是/etc/mysql/my.cnf&#xff0c;有的人的是/etc/mysql/mysql.cnf 在里最后加入 skip-grant-tables重启MySQL服务 service mysql restartShutting down MySQL… SUCCESS! Starting MySQL… SUCCESS! 重启成功 2.登…...

Webpack性能优化:构建速度与体积优化策略

一、构建速度优化 1、​​升级Webpack和Node.js​​ ​​优化效果​​&#xff1a;Webpack 4比Webpack 3构建时间降低60%-98%。​​原因​​&#xff1a; V8引擎优化&#xff08;for of替代forEach、Map/Set替代Object&#xff09;。默认使用更快的md4哈希算法。AST直接从Loa…...

STM32---外部32.768K晶振(LSE)无法起振问题

晶振是否起振主要就检查两个1、晶振与MCU是否兼容&#xff1b;2、晶振的负载电容是否匹配 目录 一、判断晶振与MCU是否兼容 二、判断负载电容是否匹配 1. 晶振负载电容&#xff08;CL&#xff09;与匹配电容&#xff08;CL1、CL2&#xff09;的关系 2. 如何选择 CL1 和 CL…...

android RelativeLayout布局

<?xml version"1.0" encoding"utf-8"?> <RelativeLayout xmlns:android"http://schemas.android.com/apk/res/android"android:layout_width"match_parent"android:layout_height"match_parent"android:gravity&…...

水泥厂自动化升级利器:Devicenet转Modbus rtu协议转换网关

在水泥厂的生产流程中&#xff0c;工业自动化网关起着至关重要的作用&#xff0c;尤其是JH-DVN-RTU疆鸿智能Devicenet转Modbus rtu协议转换网关&#xff0c;为水泥厂实现高效生产与精准控制提供了有力支持。 水泥厂设备众多&#xff0c;其中不少设备采用Devicenet协议。Devicen…...