栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 软件开发 > 后端开发 > Java

FlinkCDC获取MySQL数据库但打印不出数据或报错

Java 更新时间: 发布时间: IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

FlinkCDC获取MySQL数据库但打印不出数据或报错

flinkCDC代码:

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;



public class FlinkCDC01_DataStreamAPI {
    public static void main(String[] args) throws Exception {
        //1.创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        2.Flink-CDC将读取binlog的位置信息以状态的方式保存在CK,如果想要做到断点续传,需要从Checkpoint或者Savepoint启动程序
        2.1 开启Checkpoint,每隔5秒钟做一次CK
        //env.enableCheckpointing(5000L);
        2.2 指定CK的一致性语义
        //env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        2.3 设置任务关闭的时候保留最后一次CK数据
        //env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        2.4 指定从CK自动重启策略
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
        2.5 设置状态后端
        //env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flinkCDC"));
        2.6 设置访问HDFS的用户名
        //System.setProperty("HADOOP_USER_NAME", "atguigu");

        //3.创建Flink-MySQL-CDC的Source
        MySqlSource mySqlSource = MySqlSource.builder()
                .hostname("hadoop102")
                .port(3306)
                .username("root")
                .password("000000")
                .databaseList("test")
                .tableList("test.*")
                .deserializer(new StringDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();

        //4.使用CDC Source从MySQL读取数据
        DataStreamSource dataStreamSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MysqlSource");

        //5.打印数据
        dataStreamSource.print();

        //6.执行任务
        env.execute();

    }
}

运行,发现只有info和warn信息,没有报错,但也不打印数据

 或者,满屏报错:

 问题解决:

首先确保MySQL没问题,能读能写

打开JDK安装目录下的java.security文件

 大概700+行处,原文:

jdk.tls.disabledAlgorithms=SSLv3, TLSv1, TLSv1.1, RC4, DES, MD5withRSA, 
DH keySize < 1024, EC keySize < 224, 3DES_EDE_CBC, anon, NULL, 
include jdk.disabled.namedCurves

删去第一行中的SSLv3, TLSv1即可:

转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/979216.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号