Разбор дат в формате dd.MM.yyyy в Kafka Connect с использованием коннектора kafka-connect-spooldir - PullRequest
Новая
       6

Разбор дат в формате dd.MM.yyyy в Kafka Connect с использованием коннектора kafka-connect-spooldir

0 голосов
/ 09 февраля 2019

Я пытаюсь использовать SpoolDirCsvSourceConnector из https://github.com/jcustenborder/kafka-connect-spooldir

У меня есть следующая конфигурация для коннектора в Kafka: 

connector.class=com.github.jcustenborder.kafka.connect.spooldir.SpoolDirCsvSourceConnector
csv.first.row.as.header=true
finished.path=/csv/finished
tasks.max=1
parser.timestamp.date.formats=[dd.MM.yyyy, yyyy-MM-dd'T'HH:mm:ss, yyyy-MM-dd' 'HH:mm:ss]
key.schema={"name":"com.github.jcustenborder.kafka.connect.model.Key","type":"STRUCT","isOptional":false,"fieldSchemas":{}}
csv.separator.char=59
input.file.pattern=umsaetze_.*.csv
topic=test-csv
error.path=/csv/error
input.path=/csv/input
value.schema={"name":"com.github.jcustenborder.kafka.connect.model.Value","type":"STRUCT","isOptional":false,"fieldSchemas":{"Buchungstag":{"name":"org.apache.kafka.connect.data.Timestamp","type":"INT64","version":1,"isOptional":true},"Wertstellung":{"name":"org.apache.kafka.connect.data.Timestamp","type":"INT64","version":1,"isOptional":true},"Vorgang":{"type":"STRING","isOptional":false},"Buchungstext":{"type":"STRING","isOptional":false},"Umsatz":{"name":"org.apache.kafka.connect.data.Decimal","type":"BYTES","version":1,"parameters":{"scale":"2"},"isOptional":true}}}

схема значения следующая: 

{
  "name": "com.github.jcustenborder.kafka.connect.model.Value",
  "type": "STRUCT",
  "isOptional": false,
  "fieldSchemas": {
    "Buchungstag": {
      "name": "org.apache.kafka.connect.data.Date",
      "type": "INT32",
      "version": 1,
      "isOptional": true
    },
    "Wertstellung": {
      "name": "org.apache.kafka.connect.data.Timestamp",
      "type": "INT64",
      "version": 1,
      "isOptional": true
    },
    "Vorgang": {
      "type": "STRING",
      "isOptional": false
    },
    "Buchungstext": {
      "type": "STRING",
      "isOptional": false
    },
    "Umsatz": {
      "name": "org.apache.kafka.connect.data.Decimal",
      "type": "BYTES",
      "version": 1,
      "parameters": {
        "scale": "2"
      },
      "isOptional": true
    }
  }
}

Я пробовал Дата вместо отметок времени 

{
  "name" : "org.apache.kafka.connect.data.Date",
  "type" : "INT32",
  "version" : 1,
  "isOptional" : true
}

И отметки времени, и дата не работают для меня с тем же исключением, как в примере для полей Buchungstag и Wertstellung.Я пытался решить ее с помощью опции parser.timestamp.date.formats, но это не помогает.

Вот пример CSV, который я пытаюсь импортировать в Kafka: 

Buchungstag;Wertstellung;Vorgang;Buchungstext;Umsatz;
08.02.2019;08.02.2019;Lastschrift / Belastung;Auftraggeber: BlablaBuchungstext: Fahrschein XXXXXX Ref. U3436346/8423;-55,60;
08.02.2019;08.02.2019;Lastschrift / Belastung;Auftraggeber: Bank AGBuchungstext: 01.02.209:189,34 Ref. ZMKDVSDVS/5620;-189,34;

Я получаю следующееисключение в Kafka Connect: 

org.apache.kafka.connect.errors.ConnectException: org.apache.kafka.connect.errors.DataException: Exception thrown while parsing data for 'Buchungstag'. linenumber=2
    at com.github.jcustenborder.kafka.connect.spooldir.AbstractSourceTask.read(AbstractSourceTask.java:277)
    at com.github.jcustenborder.kafka.connect.spooldir.AbstractSourceTask.poll(AbstractSourceTask.java:144)
    ... 10 more
Caused by: org.apache.kafka.connect.errors.DataException: Could not parse '08.02.2019' to 'Date'
    at com.github.jcustenborder.kafka.connect.utils.data.Parser.parseString(Parser.java:113)
    ... 11 more
Caused by: java.lang.IllegalStateException: Could not parse '08.02.2019' to java.util.Date
    at com.google.common.base.Preconditions.checkState(Preconditions.java:588)
    ... 12 more

Есть ли у вас какие-либо идеи о том, какой должна быть схема значений для разбора дат, таких как 01.01.2001?

1 Ответ

0 голосов
/ 13 февраля 2019

Я думаю, проблема в вашем parser.timestamp.date.formats значении.Вы передаете [dd.MM.yyyy, yyyy-MM-dd'T'HH:mm:ss, yyyy-MM-dd' 'HH:mm:ss].

В конфигурации свойство (parser.timestamp.date.formats) имеет тип List.Список должен передаваться в виде строки с разделителем-запятой (,).В вашем случае это должно быть: dd.MM.yyyy,yyyy-MM-dd'T'HH:mm:ss,yyyy-MM-dd' 'HH:mm:ss.Проблема может быть с пробелами, потому что они обрезаны.

...