Flink读取mysql数据库(java)

这篇具有很好参考价值的文章主要介绍了Flink读取mysql数据库(java)。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

代码如下:

package com.weilanaoli.ruge.vlink.flink;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import java.util.logging.Level;
import java.util.logging.Logger;

class MysqlExample {
    public static void main(String[] args) throws Exception {
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("xx.xx.xx.xx")    //输入地址
                .port(3306)                 //输入端口
                .databaseList("xx")         //输入库名
                .tableList("xx.test")       //输入表名
                .username("xx")             //输入用户名
                .password("xxxx")           //输入密码
                .startupOptions(StartupOptions.initial())  //读取binlog策略,这个启动选项有五种
                .deserializer(new JsonDebeziumDeserializationSchema()) //配置不要锁表,但是数据一致性不是精准一次,会变成最少一次
                .build();
        
        //配置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // enable checkpoint
        env.enableCheckpointing(3000);
        DataStreamSource<String> dataStreamSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source");
        SingleOutputStreamOperator<Object> singleOutputStreamOperator = dataStreamSource.process(new ProcessFunction<String, Object>() {
            @Override
            public void processElement(String value, ProcessFunction<String, Object>.Context ctx, Collector<Object> out) {
                try {
                    System.out.println("processElement=====" + value);
                }catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        dataStreamSource.print("原始数据=====");
        env.execute("Print MySQL Snapshot + Binlog");
    }
}

运行结果如下:

Flink读取mysql数据库(java),Flink,数据库,flink,mysql文章来源地址https://www.toymoban.com/news/detail-655342.html

到了这里,关于Flink读取mysql数据库(java)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 【Flink】 Flink实时读取mysql数据

    准备 你需要将这两个依赖添加到 pom.xml 中 mysql mysql-connector-java 8.0.0 读取 kafka 数据 这里我依旧用的以前的 student 类,自己本地起了 kafka 然后造一些测试数据,这里我们测试发送一条数据则 sleep 10s,意味着往 kafka 中一分钟发 6 条数据。 package com.zhisheng.connectors.mysql.utils; impo

    2024年02月03日
    浏览(44)
  • 60、Flink CDC 入门介绍及Streaming ELT示例(同步Mysql数据库数据到Elasticsearch)-CDC Connector介绍及示例 (1)

    一、Flink 专栏 Flink 专栏系统介绍某一知识点,并辅以具体的示例进行说明。 1、Flink 部署系列 本部分介绍Flink的部署、配置相关基础内容。 2、Flink基础系列 本部分介绍Flink 的基础部分,比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。 3、

    2024年02月19日
    浏览(51)
  • flink任务启动抛出mysql数据库连接过多异常message from server:“Too many connections“解决办法

    1. 异常现象 2. 现象分析 2.1 mysql数据库最大默认连接数是151 2.2 已用连接数超过了最大连接数导致的异常 3. 解决办法 3.1 修改mysql最大连接数配置文件 命令行修改最大连接数(max_connections),设置最大连接数为1000。 mysql set global max_connections = 1000; 这种方式有个问题,就是设置的最

    2024年02月14日
    浏览(54)
  • 24、Flink 的table api与sql之Catalogs(java api操作数据库、表)-2

    一、Flink 专栏 Flink 专栏系统介绍某一知识点,并辅以具体的示例进行说明。 1、Flink 部署系列 本部分介绍Flink的部署、配置相关基础内容。 2、Flink基础系列 本部分介绍Flink 的基础部分,比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。 3、

    2024年02月04日
    浏览(60)
  • flink-cdc之读取mysql变化数据

    pom 代码 注意开启checkpoint 和不开启是有区别的(savepoint也可以 启动的flink指定时候 -s savepath) 不开启,如果项目重启了,会重新读取所有的数据 开启了,项目重启了额,会根据保留的信息去读取变化的数据  mysql   数据库表  增加一条数据 打印日志 op:c 是create ==FlinkCDC==

    2024年02月16日
    浏览(45)
  • Flink系列之:使用Flink CDC从数据库采集数据,设置checkpoint支持数据采集中断恢复,保证数据不丢失

    博主相关技术博客: Flink系列之:Debezium采集Mysql数据库表数据到Kafka Topic,同步kafka topic数据到StarRocks数据库 Flink系列之:使用Flink Mysql CDC基于Flink SQL同步mysql数据到StarRocks数据库

    2024年02月11日
    浏览(85)
  • Flink CDC实时同步PG数据库

    JDK:1.8 Flink:1.16.2 Scala:2.11 Hadoop:3.1.3 github地址:https://github.com/rockets0421/FlinkCDC-PG.git  1、更改配置文件postgresql.conf # 更改wal日志方式为logical wal_level = logical # minimal, replica, or logical # 更改solts最大数量(默认值为10),flink-cdc默认一张表占用一个slots max_replication_slots = 20 # m

    2024年02月13日
    浏览(69)
  • 实时Flink的数据库与Kafka集成优化案例

    在现代数据处理系统中,实时数据处理和分析是至关重要的。Apache Flink是一个流处理框架,可以用于实时数据处理和分析。在许多场景下,Flink需要与数据库和Kafka等消息系统进行集成,以实现更高效的数据处理。本文将讨论Flink与数据库和Kafka集成的优化案例,并提供实际示

    2024年02月20日
    浏览(43)
  • flink cdc 连接posgresql 数据库相关问题整理

    01 、flink posgresql cdc 前置工作 1,更改配置文件postgresql.conf wal_level是必须更改的,其它参数选着性更改,如果同步表数量超过10张建议修改为合适的值 更改配置文件postgresql.conf完成,需要重启pg服务生效,所以一般是在业务低峰期更改 2,新建用户并且给用户复制流权限 3,发

    2024年02月07日
    浏览(53)
  • flink:通过table api把文件中读取的数据写入MySQL

    当写入数据到外部数据库时,Flink 会使用 DDL 中定义的主键。如果定义了主键,则连接器将以 upsert 模式工作,否则连接器将以 append 模式工作 文件info.txt

    2024年03月15日
    浏览(49)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包