FluentdでMeCabプラグインを作成して実行すると"Unknown filter plugin 'mecab'が出て実行エラーになる

Question

前提・実現したいこと

Fluetdを用いてTwitterのStreaming APIからツイートの情報を取得し、それを形態解析するプログラムを作成しようとしています。

①rbenvをインストール
②Rubyをインストール
③Fluentdのインストール
④MeCabのインストール

発生している問題・エラーメッセージ

下記のコマンドでFluentdの起動を行うとエラーメッセージが出力されます。

fluentd -c ./fluent.twitter.mecab.conf

＜エラーメッセージ＞

2019-09-09 16:26:26 +0900 [error]: config error file="./fluent.twitter.mecab.conf" error_class=Fluent::ConfigError error="Unknown filter plugin 'mecab'. Run 'gem search -rd fluent-plugin' to find plugins"

該当のソースコード

以下はconfファイル（fluent.twitter.mecab.conf）の設定内容です。

<source>
        type twitter
        consumer_key [MY_KEY]
        consumer_secret [MY_KEY_SECRET]
        access_token [MY_TOKEN]
        access_token_secret [MY_TOKEN_SECRET]
        tag twitter
        timeline sampling
        lang ja
        output_format nest
</source>

<filter twitter>
        @type mecab
        key text
</filter>

<match twitter>
        @type stdout
</match>

<match twitter.mecab>
        type stdout
</match>

以下は./fluentd/fluent/plugin/内に作成したrbファイルの記述内容です。

module Fluent
class MeCabFileter < Filter
	Plugin.register_filter('mecab',self)
	config_param : key, : string
	config_param : tag, : string, default: "mecab"
	
	def initialize
		super
		require 'natto'
	end
	
	def configure(config)
		super
		@mecab = Natto::MeCab.new
	end
	
	def start
		super
	end
	
	def shutdown
		super
	end
	
	def filter(tag, time, record)
	end
	
	def filter_stream(tag, es)
		result_es = MultiEventStream.new
		es.each do |time, record|
		begin
			position = 0
			@mecab.parse(pre_process(record[@key])) do |mecab|
				
				length = mecab.surface.length
				next if length == 0
				
				new_record = record.clone
				new_record["mecab"] = {	"word" => mecab.surface,
										"length" => length,
										"pos" => mecab.feature.split(/\,/),
										"position" => position}
				result_es.add(time, new_record)
				
				position += length
			end
			rescue => e
				router.emit_error_event(tag, time, record, e)
			enend
			return result_es
		end
		def pre_process(text)
			# delete URL
			return text.gsub(/https?\://([\w\-]+.)+[\w-]+(/[\w-]+)*/?/,'').gsub(/RT\s*:*/,'').gsub(/@[\w]+\s*/,'')
		end
	end
end

試したこと

エラーメッセージに記載された「gem search -rd fluent-plugin」のコマンドをトライしました。mecabに関連するプラグインを探した所、以下のプラグインが見つかったので試しにそれをインストールしましたが、状況は変わりませんでした。

gem search -rld fluent-plugin
sudo gem install fluent-plugin-mecab

補足情報（FW/ツールのバージョンなど）

OS : CentOS Linux release
rbenv 1.1.2-2-g4e92322
ruby 2.3.1p112 (2016-04-26 revision 54768) [x86_64-linux]
fluentd 1.7.0
mecab of 0.996

Answer 1

プラグインの書き方が間違っています。

module Fluent
class MeCabFileter < Filter
    Plugin.register_filter('mecab',self)

ではなく、以下のように書きます。

require "natto"

module Fluent
  module Plugin
    class MeCabFilter < Filter
      Fluent::Plugin.register("mecab", self)

      def filter(tag, time, record)
        # natto でレコードを加工し、new_record を返す
      end
    end
  end
end

あと、Fluentd によって filter は最適化されるため、どうしても必要でなければ filter_stream ではなく filter を実装した方がよいです。

https://github.com/katsyoshi/fluent-plugin-mecab を参考に実装されたようですが、参考にしたプラグインの実装が古いバージョンのプラグインAPIを使用しているので、最新版のドキュメントを参照するとよいと思います。