使用fluentd mysql replicator插件将单个mysql数据复制到elasticsearch

ac1kyiln  于 2021-06-13  发布在  ElasticSearch
关注(0)|答案(1)|浏览(440)

我正在使用fluentd mysql replicator插件将我的单个mysql数据推送到elasticsearch,但是我的fluentd抛出了以下错误。internet上没有足够的可用文档,请帮助注册此错误。
我在windows本地机器上使用td-agentv3(fluentd1.10.x)版本,elasticsearch版本为7.7.1。我从td-agent命令promt以c:\opt\td-agent>fluentd-c etc\td-agent\td-agent.conf的形式运行这个配置文件

2020-12-07 13:31:42 +0530 [info]: #0 starting fluentd worker pid=12144 ppid=16648 worker=0
2020-12-07 13:31:42 +0530 [info]: #0 fluentd worker is now running worker=0
2020-12-07 13:31:43 +0530 [info]: #0 mysql_replicator: finished execution :tag=>replicator.securedb.testfluentd.${event}.${primary_key} :rows_count=>2 :elapsed_time=>0.07 sec
2020-12-07 13:31:53 +0530 [info]: #0 mysql_replicator: finished execution :tag=>replicator.securedb.testfluentd.${event}.${primary_key} :rows_count=>2 :elapsed_time=>0.26 sec
2020-12-07 13:32:03 +0530 [info]: #0 mysql_replicator: finished execution :tag=>replicator.securedb.testfluentd.${event}.${primary_key} :rows_count=>2 :elapsed_time=>0.24 sec
2020-12-07 13:32:14 +0530 [info]: #0 mysql_replicator: finished execution :tag=>replicator.securedb.testfluentd.${event}.${primary_key} :rows_count=>2 :elapsed_time=>0.25 sec
2020-12-07 13:32:24 +0530 [info]: #0 mysql_replicator: finished execution :tag=>replicator.securedb.testfluentd.${event}.${primary_key} :rows_count=>2 :elapsed_time=>0.28 sec
2020-12-07 13:32:34 +0530 [info]: #0 mysql_replicator: finished execution :tag=>replicator.securedb.testfluentd.${event}.${primary_key} :rows_count=>2 :elapsed_time=>0.23 sec
2020-12-07 13:32:44 +0530 [info]: #0 mysql_replicator: finished execution :tag=>replicator.securedb.testfluentd.${event}.${primary_key} :rows_count=>2 :elapsed_time=>0.27 sec

**2020-12-07 13:32:48 +0530 [warn]: #0 failed to flush the buffer. retry_time=0 next_retry_seconds=2020-12-07 13:32:49 +0530 chunk="5b5db3ca23998ad6be37a80c95259443" error_class=IndexError error="undefined group name reference: type_name"**

  2020-12-07 13:32:48 +0530 [warn]: #0 C:/opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluent-plugin-mysql-replicator-1.0.1/lib/fluent/plugin/out_mysql_replicator_elasticsearch.rb:65:in `[]'
  2020-12-07 13:32:48 +0530 [warn]: #0 C:/opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluent-plugin-mysql-replicator-1.0.1/lib/fluent/plugin/out_mysql_replicator_elasticsearch.rb:65:in `block in write'
  2020-12-07 13:32:48 +0530 [warn]: #0 C:/opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.11.1/lib/fluent/event.rb:325:in `each'
  2020-12-07 13:32:48 +0530 [warn]: #0 C:/opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.11.1/lib/fluent/event.rb:325:in `block in each'
  2020-12-07 13:32:48 +0530 [warn]: #0 C:/opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.11.1/lib/fluent/plugin/buffer/memory_chunk.rb:81:in `open'
  2020-12-07 13:32:48 +0530 [warn]: #0 C:/opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.11.1/lib/fluent/plugin/buffer/memory_chunk.rb:81:in `open'
  2020-12-07 13:32:48 +0530 [warn]: #0 C:/opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.11.1/lib/fluent/event.rb:324:in `each'
  2020-12-07 13:32:48 +0530 [warn]: #0 C:/opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluent-plugin-mysql-replicator-1.0.1/lib/fluent/plugin/out_mysql_replicator_elasticsearch.rb:62:in `write'
  2020-12-07 13:32:48 +0530 [warn]: #0 C:/opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.11.1/lib/fluent/plugin/output.rb:1133:in `try_flush'
  2020-12-07 13:32:48 +0530 [warn]: #0 C:/opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.11.1/lib/fluent/plugin/output.rb:1439:in `flush_thread_run'
  2020-12-07 13:32:48 +0530 [warn]: #0 C:/opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.11.1/lib/fluent/plugin/output.rb:461:in `block (2 levels) in start'
  2020-12-07 13:32:48 +0530 [warn]: #0 C:/opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.11.1/lib/fluent/plugin_helper/thread.rb:78:in `block in thread_create'
2020-12-07 13:32:49 +0530 [warn]: #0 failed to flush the buffer. retry_time=1 next_retry_seconds=2020-12-07 13:32:50 +0530 chunk="5b5db3ca23998ad6be37a80c95259443" error_class=IndexError error="undefined group name reference: type_name"
  2020-12-07 13:32:49 +0530 [warn]: #0 suppressed same stacktrace

下面是我的td代理配置文件(fluentd配置文件)。我是否需要提及elasticsearch iden name?如果是,我需要提及的地方。

<source>
  @type mysql_replicator

  # Set connection settings for replicate source.
  host 172.16.xx.xx
  port xxxxx
  username root
  password xxxxx
  database sarvendra

  # Set replicate query configuration.
  query SELECT tutorial_id, tutorial_title from testfluentd;
  primary_key tutorial_id
  interval 10s

  # Enable detect deletion event not only insert/update events. (default: yes)
  # It is useful to use `enable_delete no` that keep following recently updated record with this query.
  # `SELECT * FROM search_test WHERE DATE_ADD(updated_at, INTERVAL 5 MINUTE) > NOW();`
  enable_delete no

  # Format output tag for each events. Placeholders usage as described below.
  tag replicator.ssarvendra.testfluentd.${event}.${primary_key}
  # ${event} : the variation of row event type by insert/update/delete.
  # ${primary_key} : the value of `replicator_manager.settings.primary_key` in manager table.
</source>

<match replicator.**>
  @type mysql_replicator_elasticsearch

  # Set Elasticsearch connection.
  host localhost
  port 9200
  # You can configure to use SSL for connecting to Elasticsearch.
  # ssl true

  # Basic authentication credentials can be configured
  # username basic_auth_username
  # password basic_auth_password

  # Set Elasticsearch index, type, and unique id (primary_key) from tag.
  tag_format (?<index_name>[^\.]+)\.(?<event>[^\.]+)\.(?<primary_key>[^\.]+)$

  # Set frequency of sending bulk request to Elasticsearch node.
  flush_interval 5s

  # Set maximum retry interval (required fluentd >= 0.10.41)
  #max_retry_wait 1800

  # Queued chunks are flushed at shutdown process.
  # It's sample for td-agent. If you use Yamabiko, replace path from 'td-agent' to 'yamabiko'.
  flush_at_shutdown yes
  buffer_type file
  buffer_path /var/log/td-agent/buffer/mysql_replicator_elasticsearch
</match>
kkbh8khc

kkbh8khc1#

你应该使用以下方法 tag_format :

tag_format (?<index_name>[^\.]+)\.(?<type_name>[^\.]+)\.(?<event>[^\.]+)\.(?<primary_key>[^\.]+)$

在源代码配置中使用 tag :

tag replicator.ssarvendra._doc.${event}.${primary_key}

原因是在es7中,不再有Map类型 _doc 是唯一允许的默认类型。

相关问题