尝试使用kafka Connect在Elasticsearch中索引kafka主题

Question

我想将avro中的kafka主题索引为elasticsearch格式，但是我的时间戳字段有问题，无法由识别elasticsearch作为日期格式字段。

我对连接器使用了以下配置。

{
"name": "es-sink-barchart-10",
"config": {
"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"key.converter": "io.confluent.connect.avro.AvroConverter",
"key.converter.schema.registry.url": "http://localhost:8081",
"value.converter.schema.registry.url": "http://localhost:8081",
"connection.url": "http://localhost:9200",
"type.name":"type.name=kafka-connect",
"topics": "exchange_avro_01",
"topic.index.map": "exchange_avro_01:exchange_barchart",
"key.ignore": "true"
}
}

原始字段是bigint类型，我希望目标字段是日期类型，具有弹性搜索的任何有效格式。我定义了一个动态模板，试图通过以下方式解决它：

curl -XPUT "http://localhost:9200/_template/kafkaconnect/" -H 'Content-Type: application/json' -d'
{
"index_patterns": "exchange*",
"settings": {
"number_of_shards": 1,
"number_of_replicas": 0
},
"mappings": {
"kafka-connect": {
"dynamic_templates": [
{
"dates": {
"match_mapping_type": "long",
"match": "TIME",
"mapping": {
"type": "date",
"format": "yyyy-MM-dd HH:mm:ss"
}
}
}
]
,
"properties": {
"CLOSE": {
"type": "double"
},
.
.
.
}
}
}
}
}'

当我加载上面描述的连接器时，没有任何东西被索引到弹性搜索。

有什么帮助吗？

Answer 1

如果您的源是bigint，那么它可能是一个epoch。如果这是一个时代，那么这将不起作用：

"mapping": {
"type": "date",
"format": "yyyy-MM-dd HH:mm:ss"
}

因为你告诉Elasticsearch日期格式是yyyy-MM-dd HH:mm:ss(事实并非如此(。

因此，试着这样做(暂时省略自定义映射；先让它工作，然后再添加它(：

{
"index_patterns": "exchange*",
"settings": {
"number_of_shards": 1,
"number_of_replicas": 0
},
"mappings": {
"kafka-connect": {
"dynamic_templates": [
{
"dates": {
"match": "TIME",
"mapping": {
"type": "date"
} } } ] } } }

另请参考：https://www.elastic.co/guide/en/elasticsearch/reference/current/dynamic-field-mapping.html#date-检测

没有任何索引可用于弹性搜索。

检查Kafka Connect工作日志和Elasticsearch日志是否有任何错误。