问题描述

我试图在 sqlite 中建立一个包含两个表的数据库.我的一个表有一个时间戳列.我正在尝试实施时间戳模式来捕获数据库中的增量更改.Kafka 连接失败并出现以下错误:

I tried to set up a database with two tables in sqlite. Once of my table is having a timestamp column . I am trying to implement timestamp mode to capture incremental changes in the DB. Kafka connect is failing with the below error:

 ERROR Failed to get current time from DB using Sqlite and query 'SELECT 
CURRENT_TIMESTAMP' 
(io.confluent.connect.jdbc.dialect.SqliteDatabaseDialect:471)
java.sql.SQLException: Error parsing time stamp

Caused by: java.text.ParseException: Unparseable date: "2019-02-05 02:05:29" 
does not match (\p{Nd}++)\Q-\E(\p{Nd}++)\Q-\E(\p{Nd}++)\Q 
\E(\p{Nd}++)\Q:\E(\p{Nd}++)\Q:\E(\p{Nd}++)\Q.\E(\p{Nd}++)

非常感谢您的帮助

配置:

name=test-query-sqlite-jdbc-autoincrement 
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector 
tasks.max=1 
connection.url=jdbc:sqlite:employee.db 
query=SELECT users.id, users.name, transactions.timestamp, transactions.payment_type FROM users JOIN transactions ON (users.id = transactions.user_id) 
mode=timestamp 
timestamp.column.name=timestamp 
topic.prefix=test-joined

DDL:

CREATE TABLE transactions(id integer primary key not null,
                          payment_type text not null,
                          timestamp DATETIME DEFAULT(STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')),
                          user_id int not null, 
                          constraint fk foreign key(user_id) references users(id)
); 

CREATE TABLE users (id integer primary key not null,name text not null);

推荐答案

kafka connect jdbc 连接器很容易检测到时间戳的变化，如果 'timestamp' 列的值是 'UNIX timestamp' 的格式.

The kafka connect jdbc connector easily detects the changes in the timestamp, if the values of the 'timestamp' column are in the format of the 'UNIX timestamp'.

sqlite> CREATE TABLE transact(timestamp TIMESTAMP DEFAULT (STRFTIME('%s', 'now')) not null,
   ...> id integer primary key not null,
   ...> payment_type text not null);
sqlite>

值可以插入为:

sqlite> INSERT INTO transact(timestamp,payment_type,id) VALUES (STRFTIME('%s', 'now'),'cash',1);

然后由 kafka jdbc 源连接器检测与时间戳相关的更改，并且可以按如下方式使用:

The timestamp related changes are then detected by the kafka jdbc source connector and the same can be consumed as follows:

kafka-console-consumer  --bootstrap-server localhost:9092 --topic jdbc-transact --from-beginning
{"timestamp":1562321516,"id":2,"payment_type":"card"}
{"timestamp":1562321790,"id":1,"payment_type":"online"}

这篇关于Kafka JDBC 源连接器时间戳模式对 sqlite3 失败的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持跟版网！