Flink之JDBC Sink

这里介绍一下Flink Sink中jdbc sink的使用方法,以mysql为例,这里代码分为两种,事务和非事务

  • 非事务代码
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;import java.sql.PreparedStatement;
import java.sql.SQLException;/*** @Author: J* @Version: 1.0* @CreateTime: 2023/8/2* @Description: 测试**/
public class FlinkJdbcSink {public static void main(String[] args) throws Exception {// 构建流环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 这里使用的是自定义数据源CustomizeBean(name,age,gender,hobbit),为了方便测试,换成任何数据源都可,只要和最后的要写入的表结构匹配即可DataStreamSource<CustomizeBean> customizeSource = env.addSource(new CustomizeSource());// 构建jdbc sinkSinkFunction<CustomizeBean> jdbcSink = JdbcSink.sink("insert into t_user(`name`, `age`, `gender`, `hobbit`) values(?, ?, ?, ?)", // 数据插入sql语句new JdbcStatementBuilder<CustomizeBean>() {@Overridepublic void accept(PreparedStatement pStmt, CustomizeBean customizeBean) throws SQLException {pStmt.setString(1, customizeBean.getName());pStmt.setInt(2, customizeBean.getAge());pStmt.setString(3, customizeBean.getGender());pStmt.setString(4, customizeBean.getHobbit());}}, // 字段映射配置,这部分就和常规的java api差不多了JdbcExecutionOptions.builder().withBatchSize(10) // 批次大小,条数.withBatchIntervalMs(5000) // 批次最大等待时间.withMaxRetries(1) // 重复次数.build(), // 写入参数配置new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withDriverName("com.mysql.jdbc.Driver").withUrl("jdbc:mysql://lx01:3306/test_db?useSSL=false").withUsername("root").withPassword("password").build() // jdbc信息配置);// 添加jdbc sinkcustomizeSource.addSink(jdbcSink);env.execute();}
}
  • 事务代码
import com.mysql.cj.jdbc.MysqlXADataSource;
import org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.util.function.SerializableSupplier;import javax.sql.XADataSource;/*** @Author: J* @Version: 1.0* @CreateTime: 2023/8/2* @Description: 测试**/
public class FlinkJdbcSink {public static void main(String[] args) throws Exception {// 构建流环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 这里使用的是自定义数据源CustomizeBean(name,age,gender,hobbit),为了方便测试,换成任何数据源都可,只要和最后的要写入的表结构匹配即可DataStreamSource<CustomizeBean> customizeSource = env.addSource(new CustomizeSource());// 每20秒作为checkpoint的一个周期env.enableCheckpointing(20000);// 两次checkpoint间隔最少是10秒env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000);// 程序取消或者停止时不删除checkpointenv.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);// checkpoint必须在60秒结束,否则将丢弃env.getCheckpointConfig().setCheckpointTimeout(60000);// 同一时间只能有一个checkpointenv.getCheckpointConfig().setMaxConcurrentCheckpoints(1);// 设置EXACTLY_ONCE语义,默认就是这个env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);// checkpoint存储位置env.getCheckpointConfig().setCheckpointStorage("file:///Users/xxx/data/testData/checkpoint");// 构建ExactlyOne sink,要注意使用exactlyOnceSink需要开启checkpointSinkFunction<CustomizeBean> exactlyOneJdbcSink = JdbcSink.exactlyOnceSink("insert into t_user(`name`, `age`, `gender`, `hobbit`) values(?, ?, ?, ?)", // 数据插入sql语句(JdbcStatementBuilder<CustomizeBean>) (pStmt, customizeBean) -> {pStmt.setString(1, customizeBean.getName());pStmt.setInt(2, customizeBean.getAge());pStmt.setString(3, customizeBean.getGender());pStmt.setString(4, customizeBean.getHobbit());}, // 字段映射配置,这部分就和常规的java api差不多了JdbcExecutionOptions.builder().withMaxRetries(0) // 设置重复次数.withBatchSize(25) // 设置批次大小,数据条数.withBatchIntervalMs(1000) // 批次最大等待时间.build(),JdbcExactlyOnceOptions.builder()// 这里使用的mysql,所以要将这个参数设置为true,因为mysql不支持一个连接上开启多个事务,oracle是支持的.withTransactionPerConnection(true).build(),(SerializableSupplier<XADataSource>) () -> {// XADataSource 就是JDBC连接,不同的是它是支持分布式事务的连接MysqlXADataSource mysqlXADataSource = new MysqlXADataSource();mysqlXADataSource.setUrl("jdbc:mysql://lx01:3306/test_db?useSSL=false"); // 设置urlmysqlXADataSource.setUser("root"); // 设置用户mysqlXADataSource.setPassword("password"); // 设置密码return mysqlXADataSource;});// 添加jdbc sinkcustomizeSource.addSink(exactlyOneJdbcSink);env.execute();}
}
  • pom依赖
        <!-- 在原有的依赖中加入下面两个内容 --><!-- JDBC connector --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc</artifactId><version>${flink.version}</version></dependency><!-- mysql驱动 --><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.28</version></dependency>
  • 结果
    在这里插入图片描述
    jdbc sink的具体使用方式大概就这些内容,还是比较简单的,具体应用还要结合实际业务场景.

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

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

相关文章

设计模式概述与UML图

设计模式概述与UML图

文章目录 一、设计模式概述1. 软件设计模式的产生背景2. 软件设计模式的概念3. 学习设计模式的必要性4. 设计模式分类&#xff08;1&#xff09;创建型模式&#xff08;2&#xff09;结构型模式&#xff08;3&#xff09;行为型模式 二、UML图1. 类图概述2. 类图作用3. 类图表示…
阅读更多...
深入学习JVM —— GC垃圾回收机制

深入学习JVM —— GC垃圾回收机制

前言 前面荔枝已经梳理了有关JVM的体系结构和类加载机制&#xff0c;也详细地介绍了JVM在类加载时的双亲委派模型&#xff0c;而在这篇文章中荔枝将会比较详细地梳理有关JVM学习的另一大重点——GC垃圾回收机制的相关知识&#xff0c;重点了解的比如对象可达性的判断、四种回收…
阅读更多...
215. 数组中的第K个最大元素（快排+大根堆+小根堆）

215. 数组中的第K个最大元素（快排+大根堆+小根堆）

题目链接&#xff1a;力扣 解题思路&#xff1a; 方法一&#xff1a;基于快速排序 因为题目中只需要找到第k大的元素&#xff0c;而快速排序中&#xff0c;每一趟排序都可以确定一个最终元素的位置。 当使用快速排序对数组进行降序排序时&#xff0c;那么如果有一趟排序过程…
阅读更多...
VSCode C/C++ 分目录编译配置

VSCode C/C++ 分目录编译配置

分目录编译配置记录 launch.json文件 注释处为修改内容 {// 使用 IntelliSense 了解相关属性。 // 悬停以查看现有属性的描述。// 欲了解更多信息&#xff0c;请访问: https://go.microsoft.com/fwlink/?linkid830387"version": "0.2.0","configur…
阅读更多...
PHP8的表达式-PHP8知识详解

PHP8的表达式-PHP8知识详解

表达式是 PHP 最重要的基石。在 PHP8中&#xff0c;几乎所写的任何东西都是一个表达式。简单但却最精确的定义一个表达式的方式就是"任何有值的东西"。 最基本的表达式形式是常量和变量。当键入"$a 5"&#xff0c;即将值"5"分配给变量 $a。&quo…
阅读更多...
后端进阶之路——综述Spring Security认证，授权（一）

后端进阶之路——综述Spring Security认证，授权（一）

前言 「作者主页」&#xff1a;雪碧有白泡泡 「个人网站」&#xff1a;雪碧的个人网站 「推荐专栏」&#xff1a; ★java一站式服务 ★ ★前端炫酷代码分享 ★ ★ uniapp-从构建到提升★ ★ 从0到英雄&#xff0c;vue成神之路★ ★ 解决算法&#xff0c;一个专栏就够了★ ★ 架…
阅读更多...
【数据结构篇】手写双向链表、单向链表（超详细）

【数据结构篇】手写双向链表、单向链表（超详细）

文章目录 链表1、基本介绍2、单向链表2.1 带头节点的单向链表测试类&#xff1a;链表实现类&#xff1a; 2.2 不带头节点的单向链表2.3 练习测试类&#xff1a;链表实现类&#xff1a; 3、双向链表测试类&#xff1a;双向链表实现类&#xff1a; 4、单向环形链表**测试类**&…
阅读更多...
Gitlab CI/CD笔记-第二天-GitOps的流水线常用关键词（1）

Gitlab CI/CD笔记-第二天-GitOps的流水线常用关键词（1）

一、常用关键词 在Gitlab项目的根目录需要创建一个 .gitlab-ci.yaml的文件。 这个文件就是定义的流水线。Call :"Pipeline as code" 二、这条流水线怎么写&#xff1f; 一、掌握常用的关键词即可。 1.关键词分类 1.全局关键词 Global Keywards 2.任务关键词…
阅读更多...
Java课题笔记~ Spring 概述

Java课题笔记~ Spring 概述

Spring 框架 一、Spring 概述 1、Spring 框架是什么 Spring 是于 2003 年兴起的一个轻量级的 Java 开发框架&#xff0c;它是为了解决企业应用开发的复杂性而创建的。Spring 的核心是控制反转&#xff08;IoC&#xff09;和面向切面编程&#xff08;AOP&#xff09;。 Spring…
阅读更多...
vue 标题文字字数过长超出部分用...代替 动态显示

vue 标题文字字数过长超出部分用...代替 动态显示

效果: 浏览器最大化: 浏览器缩小: 代码: html: <div class"title overflow">{{item.name}}</div> <div class"content overflow">{{item.content}}</div> css: .overflow {/* 一定要加宽度 */width: 90%;/* 文字的大小 */he…
阅读更多...
聊聊 Docker 和 Dockerfile

聊聊 Docker 和 Dockerfile

目录 一、前言 二、了解Dockerfile 三、Dockerfile 指令 四、多阶段构建 五、Dockerfile 高级用法 六、小结 一、前言 对于开发人员来说&#xff0c;会Docker而不知道Dockerfile等于不会Docker&#xff0c;上一篇文章带大家学习了Docker的基本使用方法&#xff1a;《一文…
阅读更多...
【NLP概念源和流】 01-稀疏文档表示（第 1/20 部分）

【NLP概念源和流】 01-稀疏文档表示（第 1/20 部分）

一、介绍 自然语言处理（NLP）是计算方法的应用，不仅可以从文本中提取信息，还可以在其上对不同的应用程序进行建模。所有基于语言的文本都有系统的结构或规则，通常被称为形态学，例如“跳跃”的过去时总是“跳跃”。对于人类来说，这种形态学的理解是显而易见的。 在这篇介…
阅读更多...
设计模式-迭代器模式在Java中使用示例

设计模式-迭代器模式在Java中使用示例

场景 为开发一套销售管理系统&#xff0c;在对该系统进行分析和设计时&#xff0c;发现经常需要对系统中的商品数据、客户数据等进行遍历&#xff0c; 为了复用这些遍历代码&#xff0c;开发人员设计了一个抽象的数据集合类AbstractObjectList&#xff0c;而将存储商品和客户…
阅读更多...
MySQL正则表达式检索数据

MySQL正则表达式检索数据

目录 一、使用正则表达式进行基本字符匹配 1.使用regexp关键字 2.使用正则表达式 . 二、进行OR匹配 1.为搜索两个串之一&#xff0c;使用 | 2.匹配几个字符之一[] 3.匹配范围 4.匹配特殊字符 过滤数据允许使用匹配、比较、通配符操作来寻找数据&#xff0c;但是随…
阅读更多...
MySQL 的事件调度器

MySQL 的事件调度器

MySQL 的事件调度器可以通过以下方式进行管理&#xff1a; 1】查看事件调度器的状态 SHOW VARIABLES LIKE event_scheduler;2】启用/禁用事件调度器 SET GLOBAL event_scheduler ON;SET GLOBAL event_scheduler OFF; 注意&#xff1a;启用/禁用事件调度器需要具有 SUPE…
阅读更多...
Spring源码之XML文件中Bean标签的解析1

Spring源码之XML文件中Bean标签的解析1

读取XML文件&#xff0c;创建对象 xml文件里包含Bean的信息&#xff0c;为了避免多次IO&#xff0c;需要一次性读取xml文件中所有bean信息&#xff0c;加入到Spring工厂。 读取配置文件 new ClassPathResource("applicationContext.xml")ClassPathResource是Sprin…
阅读更多...
10倍提升效率，号称取代Elasticsearch?

10倍提升效率，号称取代Elasticsearch?

[Manticore Search](https://github.com/manticoresoftware/manticoresearch/) 是一个使用 C 开发的高性能搜索引擎&#xff0c;创建于 2017 年&#xff0c;其前身是 Sphinx Search 。Manticore Search 充分利用了 Sphinx&#xff0c;显着改进了它的功能&#xff0c;修复了数百…
阅读更多...
Charles抓包工具使用（一）（macOS）

Charles抓包工具使用（一）（macOS）

Fiddler抓包 | 竟然有这些骚操作&#xff0c;太神奇了&#xff1f; Fiddler响应拦截数据篡改&#xff0c;实现特殊场景深度测试&#xff08;一&#xff09; 利用Fiddler抓包调试工具&#xff0c;实现mock数据特殊场景深度测试&#xff08;二&#xff09; 利用Fiddler抓包调试工…
阅读更多...
Linux下TCP网络服务器与客户端通信程序入门

Linux下TCP网络服务器与客户端通信程序入门

文章目录 目标服务器与客户端通信流程TCP服务器代码TCP客户端代码 目标 实现客户端连接服务器&#xff0c;通过终端窗口发送信息给服务器端&#xff0c;服务器接收到信息后对信息数据进行回传&#xff0c;客户端读取回传信息并返回。 服务器与客户端通信流程 TCP服务器代码 …
阅读更多...
etcd

etcd

文章目录 etcd单机安装设置键值对watch操作读取键过往版本的值压缩修订版本lease租约&#xff08;过期机制&#xff09;授予租约撤销租约keepAlive续约获取租约信息 事务基于etcd实现分布式锁原生实现官方 concurrency 包实现 服务注册与发现Go 操作 Etcd 参考 etcd etcd 是一…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023