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即可:



