Kafka Elasticsearch连接器时间戳

  • Z时代
  • 2024-01-10
  • 分类：问答

我可以看到例如在这里进行了几次讨论，但是我认为由于Elasticsearch中的重大更改，解决方案已过时。

我正在尝试将我在Kafka主题中的Json中的long / epoch字段转换为通过连接器推送的Elasticsearch日期类型。

当我尝试添加动态映射时，我的Kafka连接更新失败，因为我试图将两个映射应用于字段_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)

似乎需要做的事情确实很普通，但是我看不到如何通过版本7中的此连接器将日期或时间字段输入到Elastic中？

回答：

Confluent文档指出ES 7 当前不支持 ES连接器。

根据此问题，更改type.name=kafkaconnect为type.name=_doc连接器配置可能就足够了。

以上是 Kafka Elasticsearch连接器时间戳 的全部内容， 来源链接： utcz.com/qa/410196.html

回到顶部