Попытка индексировать тему кафки в Elasticsearch с Kafka Connect - PullRequest
Новая
       50

Попытка индексировать тему кафки в Elasticsearch с Kafka Connect

0 голосов
/ 15 ноября 2018

Я хочу проиндексировать тему от kafka в формате avro до упругого поиска, но У меня проблемы с полем метки времени, который должен быть распознан asticsearch как поле формата даты.

Я использовал следующую конфигурацию для разъема. 

   {
          "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"
          },
         .
         .
         .
        }
      }

    }
  }
}'

Когда я загружаю описанный выше разъем, ничто не индексируется в эластичный поиск.

Любая помощь?

1 Ответ

0 голосов
/ 15 ноября 2018

Если ваш источник bigint, то, вероятно, это эпоха. Если это эпоха, то это не сработает: 

"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"
            } } } ] } } }

Также ref: https://www.elastic.co/guide/en/elasticsearch/reference/current/dynamic-field-mapping.html#date-detection

ничего не привязано к эластичному поиску.

Проверьте рабочий журнал Kafka Connect и журнал Elasticsearch на наличие ошибок.

...