Kafka connect possible to use custom query with bulk mode?

user_12240224 0
Viewed 2743

I'm trying to send record for every row 7 days old. This is the configuration I was working on but it doesn't work even though the query produces records on the DB server.

{
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": 1,
    "mode": "bulk",
    "connection.url": "jdbc:mysql://mysql:3300/test_db?user=root&password=password",
    "query": "SELECT * FROM test_table WHERE DATEDIFF(CURDATE(), test_table.modified) = 7;",
    "topic.prefix": "test-jdbc-",
    "poll.interval.ms": 10000
}

apache-kafka-connect apache-kafka
1 Answers

The JDBC source connector imports data from the relational database into the Apache Kafka topic by using the JDBC driver. Data is loading periodically either increment based on the timestamp or bulk load. Initially despite mode increment or bulk when you create a JDBC connector it loads all data into the topic after it will load only new or modified rows on the timestamp column.

Bulk: This mode is unfiltered and therefore not incremental at all. It will load all rows from a table on each iteration. This can be useful if you want to periodically dump an entire table where entries are eventually deleted and the downstream system can safely handle duplicates. So means you can't load the last 7 days incrementally using bulk mode

Timestamp Column: In this mode, a single column containing a modification timestamp is used to track the last time data was processed and to query only for rows that have been modified since that time. Here you can able to load incremental data. But how it works when you create the first time it will load all data available in the database table because for JDBC connector these are new data. Later it will only load new or modified data.

Now as per your requirement seems you are trying to load all data at some time interval which is going to configured "poll.interval.ms": 10000. I see your JDBC connect setting is as per definition whereas the query may not work try using the query as below. Seems JDBC connector wrap query as a table which is not working if you add where the case.

"query": "select * from (select * from test_table where  modified > now() - interval '7' day) o",

Try below setting

{
  "name": "application_name",
  "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
  "key.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "connection.url": "jdbc:mysql://mysql:3300/test_db",
  "connection.user": "root",
  "connection.password": "password",
  "connection.attempts": "1",
  "mode": "bulk",
  "validate.non.null": false,
  "query": "select * from (select * from test_table where  modified > now() - interval '7' day) o",
  "table.types": "TABLE",
  "topic.prefix": "test-jdbc-",
 "poll.interval.ms": 10000
  "schema.ignore": true,
  "key.converter.schemas.enable": "false",
  "value.converter.schemas.enable": "false"
  
}
user_25318430
Related