Штампы Kafka Elasticsearch Connector - PullRequest
Новая
       37

Штампы Kafka Elasticsearch Connector

2 голосов
/ 24 сентября 2019

Я вижу, что это обсуждалось несколько раз здесь, например , но я думаю, что решения устарели из-за серьезных изменений в Elasticsearch.

Я пытаюсь преобразоватьполе long / epoch в моем Json в моей теме Kafka для типа даты Elasticsearch, который проталкивается через соединитель.

Когда я пытаюсь добавить динамическое сопоставление, мои обновления Kafka connect завершаются неудачно, потому что я пытаюсь применить два сопоставления к полю, _doc и kafkaconnect.Это было серьезное изменение в версии 6. Я считаю, что у вас может быть только одно сопоставление на индекс. 

{
    "index_patterns": [ "depart_details" ],
  "mappings": {
    "dynamic_templates": [
      {
        "scheduled_to_date": {
          "match":   "scheduled",
          "mapping": {
            "type": "date"
          }
        }
      } 
    ]
}}

Теперь я сосредоточился на попытке перевести сообщение в источнике в соединителе, изменив поле навременная метка, время или дата. 

    "transforms.TimestampConverter.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
        "transforms.TimestampConverter.field" : "scheduled",
        "transforms.TimestampConverter.target.type": "Timestamp"

Однако любые сообщения, которые я пытаюсь отправить через этот преобразователь, терпят неудачу с 

Caused by: org.apache.kafka.connect.errors.DataException: Java class class java.util.Date does not have corresponding schema type.
    at org.apache.kafka.connect.json.JsonConverter.convertToJson(JsonConverter.java:604)
    at org.apache.kafka.connect.json.JsonConverter.convertToJson(JsonConverter.java:668)
    at org.apache.kafka.connect.json.JsonConverter.convertToJsonWithoutEnvelope(JsonConverter.java:574)
    at org.apache.kafka.connect.json.JsonConverter.fromConnectData(JsonConverter.java:324)
    at io.confluent.connect.elasticsearch.DataConverter.getPayload(DataConverter.java:181)
    at io.confluent.connect.elasticsearch.DataConverter.convertRecord(DataConverter.java:163)
    at io.confluent.connect.elasticsearch.ElasticsearchWriter.tryWriteRecord(ElasticsearchWriter.java:285)
    at io.confluent.connect.elasticsearch.ElasticsearchWriter.write(ElasticsearchWriter.java:270)
    at io.confluent.connect.elasticsearch.ElasticsearchSinkTask.put(ElasticsearchSinkTask.java:169)

Похоже, что это действительно распространенная вещь, которую нужно сделать, ноЯ не вижу, как получить поле даты или времени в Elastic через этот разъем в версии 7?

1 Ответ

1 голос
/ 24 сентября 2019

В документации Confluent говорится, что разъем ES в настоящее время не поддерживается с ES 7.

В соответствии с эта проблема может быть достаточной для изменения type.name=kafkaconnecttype.name=_doc в конфигурации вашего разъема.

...