Flink CDC MySQL同步MySQL错误记录

1、启动 Flink SQL

[appuser@whtpjfscpt01 flink-1.17.1]$ bin/sql-client.sh

在这里插入图片描述

2、新建源表

问题1：Encountered “(”
处理方法：去掉int(11)，改为int

Flink SQL> CREATE TABLE `t_user` (
>   `uid` int(11) NOT NULL AUTO_INCREMENT COMMENT 'user id',
>   `did` int(11) DEFAULT NULL COMMENT 'dept id',
>   `username` varchar(14) DEFAULT NULL,
>   `add_time` datetime DEFAULT NULL,
>   PRIMARY KEY (`uid`) NOT ENFORCED
> ) WITH (
>       'connector' = 'mysql-cdc',
>       'hostname' = '192.25.34.2',
>       'port' = '3306',
>       'username' = '*******',
>       'password' = '*******',
>       'database-name' = 'test',
>       'table-name' = 't_user'
> );
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.sql.parser.impl.ParseException: Encountered "(" at line 2, column 12.
Was expecting one of:"CONSTRAINT" ..."NOT" ..."NULL" ..."PRIMARY" ..."UNIQUE" ..."COMMENT" ..."METADATA" ...")" ..."," ..."MULTISET" ..."ARRAY" ...Flink SQL>

问题2：Encountered “AUTO_INCREMENT”
处理方法：删除AUTO_INCREMENT

Flink SQL> CREATE TABLE `t_user` (
>   `uid` int NOT NULL AUTO_INCREMENT COMMENT 'user id',
>   `did` int DEFAULT NULL COMMENT 'dept id',
>   `username` varchar(14) DEFAULT NULL,
>   `add_time` datetime DEFAULT NULL,
>   PRIMARY KEY (`uid`) NOT ENFORCED
> ) WITH (
>       'connector' = 'mysql-cdc',
>       'hostname' = '192.25.34.2',
>       'port' = '3306',
>       'username' = '*******',
>       'password' = '*******',
>       'database-name' = 'test',
>       'table-name' = 't_user'
> );
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.sql.parser.impl.ParseException: Encountered "AUTO_INCREMENT" at line 2, column 22.
Was expecting one of:"CONSTRAINT" ..."PRIMARY" ..."UNIQUE" ..."COMMENT" ..."METADATA" ...")" ..."," ..."MULTISET" ..."ARRAY" ...Flink SQL>

问题3：Encountered “DEFAULT”
处理方法：删去DEFAULT

Flink SQL> CREATE TABLE `t_user` (
>   `uid` int NOT NULL COMMENT 'user id',
>   `did` int DEFAULT NULL COMMENT 'dept id',
>   `username` varchar(14) DEFAULT NULL,
>   `add_time` datetime DEFAULT NULL,
>   PRIMARY KEY (`uid`) NOT ENFORCED
> ) WITH (
>       'connector' = 'mysql-cdc',
>       'hostname' = '192.25.34.2',
>       'port' = '3306',
>       'username' = '*******',
>       'password' = '*******',
>       'database-name' = 'test',
>       'table-name' = 't_user'
> );
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.sql.parser.impl.ParseException: Encountered "DEFAULT" at line 3, column 13.
Was expecting one of:"CONSTRAINT" ..."NOT" ..."NULL" ..."PRIMARY" ..."UNIQUE" ..."COMMENT" ..."METADATA" ...")" ..."," ..."MULTISET" ..."ARRAY" ...Flink SQL>

问题4：Unknown identifier ‘datetime’
处理方法：改用 TIMESTAMP(3)

Flink SQL> CREATE TABLE `t_user` (
>   `uid` int NOT NULL COMMENT 'user id',
>   `did` int COMMENT 'dept id',
>   `username` varchar(14) ,
>   `add_time` datetime ,
>   PRIMARY KEY (`uid`) NOT ENFORCED
> ) WITH (
>       'connector' = 'mysql-cdc',
>       'hostname' = '192.25.34.2',
>       'port' = '3306',
>       'username' = '*******',
>       'password' = '*******',
>       'database-name' = 'test',
>       'table-name' = 't_user'
> );
[ERROR] Could not execute SQL statement. Reason:
org.apache.calcite.sql.validate.SqlValidatorException: Unknown identifier 'datetime'Flink SQL>

创建成功：

Flink SQL> CREATE TABLE `t_user` (
>   `uid` int NOT NULL COMMENT 'user id',
>   `did` int COMMENT 'dept id',
>   `username` varchar(14) ,
>   `add_time` TIMESTAMP(3),
>   PRIMARY KEY (`uid`) NOT ENFORCED
> ) WITH (
>       'connector' = 'mysql-cdc',
>       'hostname' = '192.25.34.2',
>       'port' = '3306',
>       'username' = '*******',
>       'password' = '*******',
>       'database-name' = 'test',
>       'table-name' = 't_user'
> );
[INFO] Execute statement succeed.Flink SQL>

3、创建目标表

Flink SQL> CREATE TABLE `ods_t_user` (
>     `uid` int NOT NULL COMMENT 'user id',
>     `did` int COMMENT 'dept id',
>     `username` varchar(14) ,
>     `add_time` TIMESTAMP(3),
>     PRIMARY KEY (`uid`) NOT ENFORCED
>  ) WITH (
>      'connector' = 'jdbc',
>      'url' = 'jdbc:mysql://192.25.34.2:3306/demo?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC',
>      'driver' = 'com.mysql.cj.jdbc.Driver',
>      'username' = '*******',
>      'password' = '*******',
>      'table-name' = 'ods_t_user'
> );

4、将源表加载到目标表

错误1：Connector ‘mysql-cdc’ can only be used as a source. It cannot be used as a sink.

Flink SQL> insert into t_user select * from ods_t_user;
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.ValidationException: Connector 'mysql-cdc' can only be used as a source. It cannot be used as a sink.Flink SQL>

原因：方向搞反了，插入表应该是目标表

Flink SQL> insert into ods_t_user select * from t_user;
[ERROR] Could not execute SQL statement. Reason:
java.io.StreamCorruptedException: unexpected block dataFlink SQL>

错误2：unexpected block data
解决办法：
（1）更新jar包如下

[appuser@whtpjfscpt01 flink-1.17.1]$ ll lib/
total 223320
-rw-r--r-- 1 appuser appuser    196491 May 19 18:56 flink-cep-1.17.1.jar
-rw-r--r-- 1 appuser appuser    542620 May 19 18:59 flink-connector-files-1.17.1.jar
-rw-r--r-- 1 appuser appuser    266420 Sep 25 14:21 flink-connector-jdbc-3.1.1-1.17.jar
-rw-r--r-- 1 appuser appuser    345711 Sep 25 15:45 flink-connector-mysql-cdc-2.4.1.jar
-rw-r--r-- 1 appuser appuser    102472 May 19 19:02 flink-csv-1.17.1.jar
-rw-r--r-- 1 appuser appuser 135975541 May 19 19:13 flink-dist-1.17.1.jar
-rw-r--r-- 1 appuser appuser   8452171 Sep 19 10:20 flink-doris-connector-1.17-1.4.0.jar
-rw-r--r-- 1 appuser appuser    180248 May 19 19:02 flink-json-1.17.1.jar
-rw-r--r-- 1 appuser appuser  21043319 May 19 19:12 flink-scala_2.12-1.17.1.jar
-rw-r--r-- 1 appuser appuser  15407424 May 19 19:13 flink-table-api-java-uber-1.17.1.jar
-rw-r--r-- 1 appuser appuser  38191226 May 19 19:08 flink-table-planner-loader-1.17.1.jar
-rw-r--r-- 1 appuser appuser   3146210 May 19 18:56 flink-table-runtime-1.17.1.jar
-rw-r--r-- 1 appuser appuser    208006 May 17 18:07 log4j-1.2-api-2.17.1.jar
-rw-r--r-- 1 appuser appuser    301872 May 17 18:07 log4j-api-2.17.1.jar
-rw-r--r-- 1 appuser appuser   1790452 May 17 18:07 log4j-core-2.17.1.jar
-rw-r--r-- 1 appuser appuser     24279 May 17 18:07 log4j-slf4j-impl-2.17.1.jar
-rw-r--r-- 1 appuser appuser   2462364 Sep 19 11:30 mysql-connector-java-8.0.26.jar
[appuser@whtpjfscpt01 flink-1.17.1]$

（2）重启flink

[appuser@whtpjfscpt01 flink-1.17.1]$ bin/stop-cluster.sh 
Stopping taskexecutor daemon (pid: 41993) on host whtpjfscpt01.
Stopping standalonesession daemon (pid: 41597) on host whtpjfscpt01.
[appuser@whtpjfscpt01 flink-1.17.1]$ bin/start-cluster.sh 
Starting cluster.
Starting standalonesession daemon on host whtpjfscpt01.
Starting taskexecutor daemon on host whtpjfscpt01.
[appuser@whtpjfscpt01 flink-1.17.1]$ bin/sql-client.sh

（3）重新执行

Flink SQL> SET execution.checkpointing.interval = 3s;
[INFO] Execute statement succeed.Flink SQL> CREATE TABLE `t_user` (
>     `uid` int NOT NULL COMMENT 'user id',
>     `did` int COMMENT 'dept id',
>     `username` varchar(14) ,
>     `add_time` TIMESTAMP(3),
>     PRIMARY KEY (`uid`) NOT ENFORCED
> ) WITH (
>       'connector' = 'mysql-cdc',
>       'hostname' = '192.25.34.2',
>       'port' = '3306',
>       'username' = '*******',
>       'password' = '*******',
>       'database-name' = 'test',
>       'table-name' = 't_user'
>  );
[INFO] Execute statement succeed.Flink SQL> CREATE TABLE `ods_t_user` (
>     `uid` int NOT NULL COMMENT 'user id',
>     `did` int COMMENT 'dept id',
>     `username` varchar(14) ,
>     `add_time` TIMESTAMP(3),
>     PRIMARY KEY (`uid`) NOT ENFORCED
>  ) WITH (
>      'connector' = 'jdbc',
>      'url' = 'jdbc:mysql://192.25.34.2:3306/demo?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC',
>      'driver' = 'com.mysql.cj.jdbc.Driver',
>      'username' = '*******',
>      'password' = '*******',
>      'table-name' = 'ods_t_user'
> );
[INFO] Execute statement succeed.Flink SQL>

（4）成功执行

Flink SQL> insert into ods_t_user select * from t_user;
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: c2e69d061f3777c031b0acb4ec03d13a

在这里插入图片描述

错误3：无目标表
在这里插入图片描述

 CREATE TABLE demo.ods_t_user (`uid` int(11) NOT NULL AUTO_INCREMENT COMMENT 'user id',`did` int(11) DEFAULT NULL COMMENT 'dept id',`username` varchar(14) DEFAULT NULL,`add_time` datetime DEFAULT NULL,PRIMARY KEY (`uid`) 
)

在这里插入图片描述
源表添加新纪录

INSERT INTO test.t_user(did,username)values('3','test');

目标表自动同步数据
在这里插入图片描述

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

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

相关文章

Pytorch中关于forward函数的理解与用法

Pytorch中关于forward函数的理解与用法

目录 前言1. 问题所示2. 原理分析2.1 forward函数理解2.2 forward函数用法 前言 深入深度学习框架的代码，发现forward函数没有被显示调用 但代码确重写了forward函数，于是好奇是不是python的魔术方法作用 1. 问题所示 代码如下所示： cla…
阅读更多...
若依不分离+Thymeleaf select选中多个回显

若依不分离+Thymeleaf select选中多个回显

项目中遇到的场景&#xff0c;亲测实用 表单添加时&#xff0c;select选中多个&#xff0c;编辑表单时&#xff0c;select多选回显&#xff0c;如图 代码&#xff1a; // 新增代码 <label class"col-sm-3 control-label">通道&#xff1a;</label><…
阅读更多...
计算机图形学、贝塞尔曲线及绘制方法、反走样问题的解决（附完整代码）

计算机图形学、贝塞尔曲线及绘制方法、反走样问题的解决（附完整代码）

贝塞尔曲线 1. 本次作业实现的函数及简单描述&#xff08;详细代码见后&#xff09;2. 与本次作业有关的基础知识整理3. 代码描述&#xff08;详细&#xff09;4. 完整代码5. 参考文献 &#xff08;本篇为作者学习计算机图形学时根据作业所撰写的笔记&#xff0c; 如有同课程请…
阅读更多...
LabVIEW风力涡轮机的雷电流测量系统中集成高速摄像机

LabVIEW风力涡轮机的雷电流测量系统中集成高速摄像机

LabVIEW风力涡轮机的雷电流测量系统中集成高速摄像机 随着全球风电装机容量的快速增长&#xff0c;雷电活动对风力发电机组造成的损害受到更多关注&#xff0c;特别是在雷电活动强烈的地区。在冬季闪电期间&#xff0c;风力涡轮机等高层结构会受到向上的雷击。众所周知&#x…
阅读更多...
HelloWorld显示Go语言交叉编译的强大20230926

HelloWorld显示Go语言交叉编译的强大20230926

环境介绍 开发环境:windows 10 IDE:goland 实现的目标: 在windows10下编译go,分别在linux centos6和linux centos8上进行运行 具体流程 1.在windows10上建立项目 a. 打开GoLand&#xff0c;选择New Project。 b. 为项目取一个名称&#xff0c;例如HelloWorld&#xff0c…
阅读更多...
Acwing 837. 连通块中点的数量

Acwing 837. 连通块中点的数量

Acwing 837. 连通块中点的数量 题目描述思路讲解代码展示 题目描述 思路讲解 大家看y总这段代码时要注意&#xff0c;在C操作时&#xff0c;y总先把a&#xff0c;b的根结点取出来了&#xff1a;a find(a), b find(b);&#xff0c;因此接下来是先将集合a接到集合b下再把a的连通…
阅读更多...
Android修行手册 - Activity 在 Java 和 Kotlin 中怎么写构造参数

Android修行手册 - Activity 在 Java 和 Kotlin 中怎么写构造参数

点击跳转>Unity3D特效百例点击跳转>案例项目实战源码点击跳转>游戏脚本-辅助自动化点击跳转>Android控件全解手册点击跳转>Scratch编程案例点击跳转>软考全系列 &#x1f449;关于作者 专注于Android/Unity和各种游戏开发技巧&#xff0c;以及各种资源分享&…
阅读更多...
构建捡垃圾机器人的 ROS 2 项目

构建捡垃圾机器人的 ROS 2 项目

一、说明 本系列是关于学习如何使用 ROS2、Docker 和 Github 设计、设置和维护机器人项目。 先决条件 — ROS2 软件包的基本知识、实现发布者、订阅者、操作并连接它们。 我们之前在 ROS2 中了解了不同的部分。但是&#xff0c;在我们转向实际的基于硬件的项目之前&#xff0c;…
阅读更多...
阿里云ECS服务器无法发送邮件问题解决方案

阿里云ECS服务器无法发送邮件问题解决方案

这篇文章分享一下自己把项目部署在阿里云ECS上之后&#xff0c;登录邮件提醒时的邮件发送失败问题&#xff0c;无法连接发送邮箱的服务器。 博主使用的springboot提供的发送邮件服务&#xff0c;如下所示&#xff0c;为了实现异步的效果&#xff0c;新开了一个线程来发送邮件。…
阅读更多...
基于 SpringBoot 2.7.x 使用最新的 Elasticsearch Java API Client 之 ElasticsearchClient

基于 SpringBoot 2.7.x 使用最新的 Elasticsearch Java API Client 之 ElasticsearchClient

1. 从 RestHighLevelClient 到 ElasticsearchClient 从 Java Rest Client 7.15.0 版本开始&#xff0c;Elasticsearch 官方决定将 RestHighLevelClient 标记为废弃的&#xff0c;并推荐使用新的 Java API Client&#xff0c;即 ElasticsearchClient. 为什么要将 RestHighLevelC…
阅读更多...
SuffixArray练习题

SuffixArray练习题

SuffixArray练习题 🍉题目 import java.util.Arrays;class SuffixArray {//LCP:Longest common prefix/*字符串后缀，指从字符串某个* 位置开始到字符串末尾的字串，原串和空串也是后缀* Create the LCP array from the suffix array* 从后缀数组创建LCP数组* @param s the…
阅读更多...
Windows的批处理——获取系统时间、生成当天日期日志

Windows的批处理——获取系统时间、生成当天日期日志

Windows批处理基础https://coffeemilk.blog.csdn.net/article/details/132118351 一、Windows批处理的日期时间 在我们进行软件开发的过程中&#xff0c;有时候会使用到一些批处理命令&#xff0c;其中就涉及到获取系统日期、时间来进行一些逻辑的判断处理&#xff1b;那么我们…
阅读更多...
Tomcat启动后的日志输出为乱码

Tomcat启动后的日志输出为乱码

天行健&#xff0c;君子以自强不息&#xff1b;地势坤&#xff0c;君子以厚德载物。 每个人都有惰性&#xff0c;但不断学习是好好生活的根本&#xff0c;共勉&#xff01; 文章均为学习整理笔记&#xff0c;分享记录为主&#xff0c;如有错误请指正&#xff0c;共同学习进步。…
阅读更多...
[Linux] 4.常用初级指令

[Linux] 4.常用初级指令

pwd&#xff1a;显示当前文件路径 ls:列出当前文件夹下有哪些文件 mkdir空格文件名&#xff1a;创建一个新的文件夹 cd空格文件夹名&#xff1a;进入文件夹 cd..&#xff1a;退到上一层文件夹 ls -a&#xff1a;把所有文件夹列出来 .代表当前文件夹 ..代表上层文件夹 用…
阅读更多...
朴素贝叶斯分类（下）：数据挖掘十大算法之一

朴素贝叶斯分类（下）：数据挖掘十大算法之一

⭐️⭐️⭐️⭐️⭐️欢迎来到我的博客⭐️⭐️⭐️⭐️⭐️ 🐴作者：秋无之地 🐴简介：CSDN爬虫、后端、大数据领域创作者。目前从事python爬虫、后端和大数据等相关工作，主要擅长领域有：爬虫、后端、大数据开发、数据分析等。 🐴欢迎小伙伴们点赞👍🏻、收藏⭐️、…
阅读更多...
探索ClickHouse——连接Kafka和Clickhouse

探索ClickHouse——连接Kafka和Clickhouse

安装Kafka 新增用户 sudo adduser kafka sudo adduser kafka sudo su -l kafka安装JDK sudo apt-get install openjdk-8-jre下载解压kafka 可以从https://downloads.apache.org/kafka/下找到希望安装的版本。需要注意的是&#xff0c;不要下载路径包含src的包&#xff0c;否…
阅读更多...
最新ChatGPT网站系统源码+支持GPT4.0+支持AI绘画Midjourney绘画+支持国内全AI模型

最新ChatGPT网站系统源码+支持GPT4.0+支持AI绘画Midjourney绘画+支持国内全AI模型

一、SparkAI创作系统 SparkAi系统是基于很火的GPT提问进行开发的Ai智能问答系统。本期针对源码系统整体测试下来非常完美&#xff0c;可以说SparkAi是目前国内一款的ChatGPT对接OpenAI软件系统。那么如何搭建部署AI创作ChatGPT系统&#xff1f;小编这里写一个详细图文教程吧&a…
阅读更多...
AWS】在EC2上创建root用户，并使用root用户登录

AWS】在EC2上创建root用户，并使用root用户登录

最近有项目需要使用AWS的EC2服务器&#xff1b; 在创建服务器实例之后发现&#xff0c;没有root用户&#xff0c;仔细阅读AWS EC2文档&#xff0c;发现默认是ec2-user用户&#xff1b; 那我们需要创建一个root用户 1.创建 root 用户 注意&#xff1a;必须要要在ec2-user用户下…
阅读更多...
CCF CSP认证 历年题目自练Day18

CCF CSP认证 历年题目自练Day18

CCF CSP认证 历年题目自练Day18 题目一 试题编号&#xff1a; 201809-1 试题名称&#xff1a; 卖菜 时间限制&#xff1a; 1.0s 内存限制&#xff1a; 256.0MB 问题描述&#xff1a; 问题描述 　　在一条街上有n个卖菜的商店&#xff0c;按1至n的顺序排成一排&#xff0c;这…
阅读更多...
Apollo自动驾驶系统概述（文末参与活动赠送百度周边）

Apollo自动驾驶系统概述（文末参与活动赠送百度周边）

前言 「作者主页」&#xff1a;雪碧有白泡泡 「个人网站」&#xff1a;雪碧的个人网站 「推荐专栏」&#xff1a; ★java一站式服务 ★ ★ React从入门到精通★ ★前端炫酷代码分享 ★ ★ 从0到英雄&#xff0c;vue成神之路★ ★ uniapp-从构建到提升★ ★ 从0到英雄&#xff…
阅读更多...
最新文章

Copyright @ 2022~2023