Kafka Connect MySQL для Redshift - Даты Причина Ошибка - PullRequest
Новая
       6

Kafka Connect MySQL для Redshift - Даты Причина Ошибка

0 голосов
/ 19 октября 2018

Я получаю эту ошибку при попытке потоковой передачи таблицы RDS MySQL в Redshift: Ошибка преобразования данных, недопустимый тип для параметра

Поле проблемы: DATETIME в MySQL иtimestamp without time zone в Redshift (то же самое происходит для timestamp with time zone).Примечание: конвейер работал нормально, пока я не заполнил поле даты.

Мы используем Debezium в качестве источника Kafka Connect для передачи данных из RDS в Kafka.И разъем приемника JDBC с драйвером JDBC Redshift для приемника.

Также ... Я могу передавать данные, если сделаю поле Redshift varchar или bigint.Когда я делаю это, я вижу, что данные представляются как целое число эпохи Unix в мс.Но нам бы очень понравилась временная метка!

Сообщение об ошибке в контексте: 

2018-10-18 22:48:32,972 DEBUG  ||  INSERT sql: INSERT INTO "funschema"."test_table"("user_id","subscription_code","source","receipt","starts_on") VALUES(?,?,?,?,?)   [io.confluent.connect.jdbc.sink.BufferedRecords]
2018-10-18 22:48:32,987 WARN   ||  Write of 28 records failed, remainingRetries=7   [io.confluent.connect.jdbc.sink.JdbcSinkTask]
java.sql.BatchUpdateException: [Amazon][JDBC](10120) Error converting data, invalid type for parameter: 5.
    at com.amazon.jdbc.common.SStatement.createBatchUpdateException(Unknown Source)
    at com.amazon.jdbc.common.SStatement.access$100(Unknown Source)
    at com.amazon.jdbc.common.SStatement$BatchExecutionContext.createBatchUpdateException(Unknown Source)
    at com.amazon.jdbc.common.SStatement$BatchExecutionContext.createResults(Unknown Source)
    at com.amazon.jdbc.common.SStatement$BatchExecutionContext.doProcess(Unknown Source)
    at com.amazon.jdbc.common.SStatement$BatchExecutionContext.processInt(Unknown Source)
    at com.amazon.jdbc.common.SStatement.processBatchResults(Unknown Source)
    at com.amazon.jdbc.common.SPreparedStatement.executeBatch(Unknown Source)
    at io.confluent.connect.jdbc.sink.BufferedRecords.flush(BufferedRecords.java:138)
    at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:66)
    at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:75)

Спасибо,

Том

...