Confluent: ERROR Не удалось выполнить запрос для таблицы TimestampIncrementingTableQuerier mysql-jdbc

Question

Я пытаюсь использовать метку времени режима с MySQL, с ограниченными строками, так как размер моей таблицы составляет 2,6 ГБ.

Вот свойства соединителя, которые я использую:

{
        "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-"
        }
}

Iкак показано ниже:

INFO TimestampIncrementingTableQuerier {table = null, query = 'SELECT matriid, DateUpdated from users.employee WHERE date (DateUpdated)> =' 2018-11-28 '', topicPrefix = 'mysql-prod-kot- ', incrementingColumn =' ', timestampColumns = [DateUpdated]} подготовленный SQL-запрос: SELECT matriid, DateUpdated from users.employee WHERE date (DateUpdated)> =' 2018-11-28 'WHERE DateUpdated>?И DateUpdated <?ORDER BY <code>DateUpdated ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier: 161) [2018-11-29 17: 29: 00,981] ОШИБКА Не удалось выполнить запрос для таблицы TimestampIncrementingTableQuerier {table = null, query = 'SELECTmatriid, DateUpdated from users.employee WHERE date (DateUpdated)> = '2018-11-28' ', topicPrefix =' mysql-prod-kot- ', incrementingColumn =' ', timestampColumns = [DateUpdated]}: {} (io.confluent.connect.jdbc.source.JdbcSourceTask: 328) java.sql.SQLSyntaxErrorException: в синтаксисе SQL имеется ошибка;проверьте руководство, соответствующее вашей версии сервера MySQL, чтобы найти правильный синтаксис для использования рядом с 'WHERE DateUpdated>' 1970-01-01 00: 00: 00.0 'AND DateUpdated <' 2018-11-29 17 'в строке1 </p>

Answer 1

Это происходит потому, что вы пытаетесь использовать "mode": "timestamp" и query.TimestampIncrementingTableQuerier добавляет к запросу предложение WHERE, которое конфликтует с существующими предложениями WHERE в query.

Документация по исходному соединителю JDBC понятна по этому вопросу:

query

Если указано, запрос для выбора нового илиобновленные строки.Используйте этот параметр, если вы хотите объединить таблицы, выбрать подмножества столбцов в таблице или отфильтровать данные.При использовании этот соединитель будет копировать данные только по этому запросу - копирование всей таблицы будет отключено.Различные режимы запросов все еще могут использоваться для инкрементных обновлений, но для правильного построения инкрементного запроса должна быть возможность добавить к этому запросу предложение WHERE (то есть нельзя использовать предложения 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, когда вы такжев запросе указан один из них, что приводит к недопустимому SQL

Per docs для соединителя источника JDBC:

для правильного построения инкрементного запроса,должна быть возможность добавить предложение WHERE к этому запросу (т. е. нельзя использовать предложения WHERE ).Если вы используете предложение WHERE, оно само должно обрабатывать инкрементные запросы.