如何在 Kafka Connect JDBC Source 连接器中添加显式 WHERE 子句答案

【问题标题】：How to add explicit WHERE clause in Kafka Connect JDBC Source connector如何在 Kafka Connect JDBC Source 连接器中添加显式 WHERE 子句
【发布时间】：2019-10-30 22:22:24
【问题描述】：

我正在使用 kafka 连接从 DB2 到 kafka 主题的源数据，并且我正在配置 sql 查询以从 DB2 读取数据，下面是查询

SELECT SEQ_I AS error_id, TRIM(SEND_I) AS sca , to_char(CREATE_TS,'YYYY-MM-DD HH24:MI:SS.FF3') AS create_timestamp, CREATE_TS, TRIM(ERR_MSG) AS error_message , CASE substr(ERR_MSG,1,locate('-',ERR_MSG)-1) WHEN 'WARNING' THEN 'W' WHEN 'SUSPENDED' THEN 'F' END ERROR_TYPE FROM INTCHG_ERROR_DIR WHERE TRAN_I ='503' AND PRCS_N = 'GLOBAL'

我正在使用设置 "timestamp.column.name": "CREATE_TS" 这里问题出在查询中，它们已经是 WHERE 子句，并且 kafka connect 尝试添加另一个带有时间戳列的 where 子句，它正在创建问题，另一个问题是如果我删除 where 子句来自下面的 sql 子句

SELECT SEQ_I AS error_id, TRIM(SEND_I) AS sca , to_char(CREATE_TS,'YYYY-MM-DD HH24:MI:SS.FF3') AS create_timestamp, CREATE_TS, TRIM(ERR_MSG) AS error_message , CASE substr(ERR_MSG,1,locate('-',ERR_MSG)-1) WHEN 'WARNING' THEN 'W' WHEN 'SUSPENDED' THEN 'F' END ERROR_TYPE FROM INTCHG_ERROR_DIR

然后出现 substr 错误，如下所示

SQL Error [22011]: THE SECOND OR THIRD ARGUMENT OF THE SUBSTR OR SUBSTRING FUNCTION IS OUT OF RANGE. SQLCODE=-138, SQLSTATE=22011, DRIVER=4.19.26

任何人都可以就这两个问题提出建议吗，我卡在这一点上。

【问题讨论】：

标签： jdbc apache-kafka db2 apache-kafka-connect

【解决方案1】：

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

JDBC source connector docs 对此很清楚：

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 SEQ_I AS error_id, TRIM(SEND_I) AS sca , to_char(CREATE_TS,'YYYY-MM-DD HH24:MI:SS.FF3') AS create_timestamp, CREATE_TS, TRIM(ERR_MSG) AS error_message , CASE substr(ERR_MSG,1,locate('-',ERR_MSG)-1) WHEN 'WARNING' THEN 'W' WHEN 'SUSPENDED' THEN 'F' END ERROR_TYPE FROM INTCHG_ERROR_DIR WHERE TRAN_I ='503' AND PRCS_N = 'GLOBAL') o"

【讨论】：