当前位置: 首页 > 面试题库 >

尝试使用Kafka Connect在Elasticsearch中为Kafka主题编制索引

公良琛
2023-03-14
问题内容

我想将afro中的主题从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类型,我希望目标字段是具有Elasticsearch的任何有效格式的日期类型。我定义了一个动态模板,尝试通过以下方式解决它:

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

    }
  }
}'

当我加载上述连接器时,没有任何东西索引到Elasticsearch。

有什么帮助吗?


问题答案:

如果您的来源是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"
            } } } ] } } }

另请参阅:https : //www.elastic.co/guide/zh-
CN/elasticsearch/reference/current/dynamic-field-mapping.html#date-
detection

没有东西被索引到elasticsearch。

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



 类似资料:
  • 问题内容: 我刚开始使用Elasticsearch。我们的要求是我们需要索引成千上万个PDF文件,而我很难让其中一个成功地索引。 安装了附件类型插件并获得响应:。 遵循了“操作中的附件类型”教程,但是该过程挂起,并且 我不知道如何解释错误消息 。还尝试了悬挂在同一个地方的要点。 更多细节: 该包含一个嵌入式的Base64 PDF文件(按说明)。该文件的第一行 似乎 正确(无论如何对我来说):… 我

  • 我想在kafka主题中搜索特定的消息,我找到的唯一解决方案是使用 有没有一个有效的方法做这件事? 有没有一种方法可以用一个特定的偏移量来限制使用者,即从一开始读到in达到特定的偏移量?

  • 问题内容: 我有一个logstash输入设置为 我需要将主题提供给Elasticsearch中的两个不同的索引。任何人都可以帮助我如何为此类任务设置输出。目前,我只能设置 我需要在同一elasticsearch例如两个指标说和,这将在未来对信息供给和 问题答案: 首先,您需要添加到输入中才能知道消息来自哪个主题 然后,您有两个选择,都涉及条件逻辑。首先是通过引入一个过滤器来根据主题名称添加正确的索

  • 问题内容: 我有一个大约2亿个文档的ElasticSearch索引,总索引大小为90Gb。 我更改了映射,所以我希望ElasticSearch重新索引所有文档。 我编写了一个脚本,该脚本创建一个新索引(具有新映射),然后遍历旧索引中的所有文档,然后将其放入新索引中。 它似乎可以工作,但是问题是它的运行速度非常慢。它从两天前以300个文档/分钟开始,现在的速度是150个文档/分钟。 该脚本在elas

  • 问题内容: 我正在使用Django 1.4,Haystack 2 beta和ElasticSearch .20运行安装程序。我的数据库是PostgreSQL 9.1,它有几百万条记录。当我尝试使用haystack / elasticsearch索引所有数据时,该过程超时,并且收到一条消息,提示“已杀死”。到目前为止,我已经注意到以下几点: 我确实获得了要建立索引的文档数量,因此没有出现“ 0个要建

  • 我已经在本地安装了kafka(目前没有集群/模式注册),并尝试生成一个Avro主题,下面是与该主题相关的模式。 我想创建一个简单的来根据上述模式创建一些数据并将其发布到kafka。考虑创建转换为的示例数据,然后将其更改为然后发布。 然后,如下所示: 现在可以< code >手动将模式附加到此avro主题。 这可以通过使用而不是使用来实现吗?这是用于批处理,而不是。