Confluent:ERROR无法运行表TimestampIncrementingTableQuerier-mysql-

Question

我正尝试在MySQL中使用模式时间戳，因为我的表大小为2.6 GB，所以行数有限。

以下是我正在使用的连接器属性：

{
"name": "jdbc_source_mysql_registration_query",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"key.converter": "io.confluent.connect.avro.AvroConverter",
"key.converter.schema.registry.url": "http://localhost:8081",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url": "http://localhost:8081",
"connection.url": "jdbc:mysql://localhost:3310/users?zeroDateTimeBehavior=ROUND&useCursorFetch=true&defaultFetchSize=1000&user=kotesh&password=kotesh",
"query": "SELECT matriid,DateUpdated  from users.employee WHERE date(DateUpdated)>='2018-11-28' ",
"mode": "timestamp",
"timestamp.column.name": "DateUpdated",
"validate.non.null": "false",
"topic.prefix": "mysql-prod-kot-"
}
}

我得到如下：

INFO TimestampIncrementingTableQuerier｛table=null，query='SELECTmatrixid，从users.employee WHERE更新的日期date(DateUpdated(>='2018-11-28''，主题前缀='mysql-prod-kot-'，incrementingColumn=''，timestampColumns=[DateUpdated]}已准备好SQLquery：SELECT matrixid，从users.employee WHERE更新的日期date(DateUpdated(>='2018-11-28'哪里有DateUpdated>？以及DateUpdated<ccd3 ASC订购(io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:161(〔2018-11-29 17:29:00981〕错误无法运行表TimestampIncrementingTableQuerier的查询｛table=null，query='SELECTmatrixid，从users.employee WHERE更新的日期date(DateUpdated(>='2018-11-28''，主题前缀='mysql-prod-kot-'，incrementingColumn="，timestampColumns=[DateUpdated]｝：｛｝(io.confluent.connect.jdbc.source.JdbcSourceTask:328(java.sql.SQLSyntaxErrorException：您的sql语法中有一个错误；查看与MySQL服务器版本相对应的手册有关在'WHEREDateUpdated>'1970-01-01附近使用的正确语法00:00:00.0'与DateUpdated<'2018-11-29 17'在1号线

Answer 1

发生这种情况是因为您试图同时使用"mode": "timestamp"和query。TimestampIncrementingTableQuerier将WHERE子句附加到与query中现有WHERE子句冲突的查询中。

JDBC源连接器文档对此很清楚：

query

如果指定，则执行查询以选择新行或更新的行。使用该设置如果要联接表，请在中选择列的子集表或过滤器数据。如果使用，此连接器将仅复制数据使用此查询--将禁用整表复制。不同的查询模式仍然可以用于增量更新，但为了正确构造增量查询，必须能够将WHERE子句附加到此查询(即使用(如果使用WHERE子句，它必须处理增量查询本身。

作为一种变通方法，您可以将查询修改为(取决于您使用的SQL风格(

SELECT * FROM ( SELECT * FROM table WHERE ...)

或

WITH a AS
SELECT * FROM b
WHERE ...
SELECT * FROM a

例如，在您的情况下，查询应该是

"query":"SELECT * FROM (SELECT matriid,DateUpdated  from users.employee WHERE date(DateUpdated)>='2018-11-28') o"

Answer 2

错误如图所示：

java.sql.SQLSyntaxErrorException: You have an error in your SQL syntax; 
check the manual that corresponds to your MySQL server version for the right syntax to use near 
'WHERE `DateUpdated` > '1970-01-01 00:00:00.0' AND `DateUpdated` < '2018-11-29 17' at line 1

这是因为您使用的是query，但也使用了"mode": "timestamp"，因此当您在查询中也指定了WHERE子句时，连接器会尝试附加它自己的CCD_15语句，这会导致SQL 无效

JDBC源连接器的每个文档：

为了正确构造增量查询，必须能够在此查询中附加WHERE子句(即不能使用WHERE子句(。如果使用WHERE子句，它必须自己处理增量查询。