Logstash Kafka連携の設計｜input/outputプラグインの設定とElasticsearch配送・重複と再送を実装視点で解説

LogstashとKafkaをつなぐ構成でつまずく原因は、設定項目の多さではなく、どちらの向きにつなぐのかを決めないまま書き始めることにあります。同じkafkaという名前のプラグインでも、inputに置けばLogstashはKafkaの消費者になり、outputに置けば生産者になる。役割が正反対なので、決めるべき設定も、詰まったときに起きることも変わります。この記事では、向きの決定から始めて、Elasticsearchへの配送設計と再送・重複の扱いまでを設計の順番どおりに追います。Apache Kafka本体の構成要素と役割は別記事で扱っているため、ここではつなぎ手側の判断に絞りました。

示す既定値と版番号は、2026年8月30日にElastic公式ドキュメントで実測した値です。同日時点で logstash-integration-kafka は v12.1.0（2026年3月18日公開）が案内されており、Kafka連携のinputプラグインとoutputプラグインはこの一つの統合パッケージに同梱されていました。Logstash本体は 9.5系（同日時点の最新は 9.5.2）です。

まとめ｜Logstash Kafka連携で先に決める三つの設計判断

先に結論を置きます。決めるべきは、つなぐ向き・詰まりの逃げ場・重複の潰し方の三つです。向きはKafkaを取り込み元にするか送り先にするかで、これが決まればプラグイン設定の大半は自動的に決まります。

詰まりの逃げ場は、Logstash内部の永続キューではなくKafkaのトピック側に置くのが素直な設計になります。Kafkaを前段に挟む構成なら、Elasticsearchが遅れてもLogstashが読む速度を落とすだけで済み、保持期間の範囲内であればデータは消えません。ここを内部キューで受け止めようとすると、Logstashのディスクが先に埋まります。

重複の潰し方は配送先で決めます。Kafkaの取り込みは少なくとも1回の到達しか保証しないため、再送時の重複は前提として設計します。Elasticsearch側で document_id にKafkaのトピック・パーティション・オフセットを組み合わせた値を渡し、同じイベントが二度届いても上書きになる形にしておくのが定石です。以下、この順に設計手順を追います。

LogstashのKafka連携はinputとoutputで役割が正反対になる

設定ファイルの構造は input と filter と output の三段で、Kafkaはこのうち両端どちらにも置けます。同じ統合パッケージの中に二つのプラグインが入っているため、名前だけを見て混同しやすい部分です。

input側は消費者グループとしてKafkaからイベントを取り込む

inputに置く構成は、すでにKafkaへ集約されたログやイベントをLogstashが読み出し、変換してElasticsearchなどへ届ける形になります。実務で多いのはこちらです。Logstashは通常のKafkaコンシューマーとして振る舞い、group_id で指定した消費者グループに参加して、パーティションの割り当てを受けます。

この構成の利点は、Logstashを増減させても取りこぼしが起きにくいことにあります。同じ group_id で複数のLogstashを立ち上げれば、Kafka側が自動でパーティションを配り直すためです。逆に、別系統の処理を足すつもりで同じ group_id を使い回すと、イベントが両方には届かず取り合いになります。用途ごとにグループを分けてください。

output側はプロデューサーとしてKafkaのトピックへ書き戻す

outputに置く構成は、Beatsやファイル、syslogから受けたイベントをLogstashが整形し、Kafkaのトピックへ書き込む形です。後段に複数の消費者がいる場合や、配送先の障害を吸収する緩衝地帯が欲しい場合に選びます。topic_id だけが必須設定で、他は既定値のままでも動作します。

ただし既定値のままでは欠損に弱い構成になります。2026年8月30日時点のドキュメントでは acks の既定は 1 で、リーダーのブローカーが受け取った時点で成功とみなす設定です。リーダーが複製前に落ちるとそのイベントは失われます。欠損を嫌う経路では acks を all にし、送信側の再試行とあわせて設計してください。

Kafka inputプラグインの設定はグループとスレッド数から決める

input側の設定は項目数こそ多いものの、実際に判断が要るのは4つほどです。残りは既定値のままで動きます。以下は取り込み経路の典型的な設定例です。

input {
  kafka {
    bootstrap_servers => "kafka-1:9092,kafka-2:9092,kafka-3:9092"
    topics => ["app-logs"]
    group_id => "logstash-app-logs"
    consumer_threads => 3
    auto_offset_reset => "latest"
    enable_auto_commit => false
    decorate_events => "basic"
    codec => "json"
  }
}

group_idとconsumer_threadsはパーティション数に合わせる

既定値は group_id が logstash 、consumer_threads が 1 です。前者を既定のまま複数系統で動かすと前述の取り合いが起きるため、用途が分かるグループ名を必ず付けます。

スレッド数はパーティション数から逆算します。Elastic公式ドキュメントには、パーティション数と同じスレッド数を持つのが理想であり、スレッドがパーティション数を超えると余った分は待機状態になり得る旨の記述があります。つまり consumer_threads を増やしても、パーティションが少なければ並列度は上がりません。取り込みが追いつかないときに最初に見るのはスレッド数ではなく、トピック側のパーティション数です。

Logstashを複数台に分ける場合は、台数とスレッド数の積がパーティション数を超えないように配ります。超えた分は動いているのに何もしないコンシューマーになり、リバランスのたびに割り当てが揺れるだけの存在になってしまう。

auto_offset_resetとオフセット管理で取りこぼしが変わる

オフセットの扱いは、障害時にどこから読み直すかを決める設定です。auto_offset_reset はコミット済みのオフセットが見つからないときの挙動を決め、latest なら最新から、earliest なら保持されている先頭から読み始めます。監視ログのように過去分を取り戻す必要がない経路は前者、監査用途で欠落を許さない経路は後者を選びます。

もう一つの分岐が enable_auto_commit です。既定は true で、一定間隔でオフセットが自動的にコミットされます。この設定だと、Logstashが処理し終える前にコミットが走った直後にプロセスが落ちた場合、そのイベントは二度と読み直されません。取りこぼしを嫌う経路では false にして、パイプラインが処理を終えた区切りでコミットさせる構成にします。その代わり同じイベントが二度届く可能性が上がるため、後述の重複排除とセットで判断してください。

decorate_eventsでKafkaのメタデータを後段に引き渡す

既定の decorate_events は none で、Kafka由来の情報はイベントに付きません。basic にすると [@metadata][kafka][topic] や [@metadata][kafka][partition] 、[@metadata][kafka][offset] といったメタデータが載り、output側から参照できるようになります。

この設定は、インデックス名をトピックごとに切り替えたい場合と、重複排除のキーを組み立てたい場合に効きます。逆に言えば、この二つをやらないなら既定のままで構いません。@metadata 配下のフィールドはElasticsearchへは送られないため、付けてもドキュメントは太らない点も押さえておきます。

Kafkaから流れてくるのが構造化されていない生ログの場合、ここで codec をJSONにしても意味がありません。フィルター側で分解する必要があり、その手法はLogstashのgrokフィルターの使い方（構文とカスタムパターン）にまとめています。

Elasticsearchへの配送はbulkの単位と重複排除の設計で決まる

Kafkaから読んだイベントをElasticsearchへ届ける部分は、Logstash側というより配送先側の設計になります。2026年8月30日時点の logstash-output-elasticsearch は v12.1.2（2026年2月11日公開）でした。

Elasticsearch outputはbulk APIでまとめて書き込む

このプラグインはイベントを1件ずつ送らず、bulk APIでまとめて投入します。公式ドキュメントには、バッチが20MBを超える場合に複数のリクエストへ分割する旨の記載があります。したがって調整の対象になるのはHTTPの発行回数ではなく、パイプライン側のバッチ件数とワーカー数です。

bulkは部分的に失敗しうる点に注意が要ります。100件を送って3件だけがマッピング違反で弾かれる状況では、リクエスト自体は成功として返る。bulk APIのNDJSON形式と部分失敗の扱いを押さえたうえで、弾かれたイベントをデッドレターキューへ落とす設定を入れておくと、原因追跡が現実的になります。

document_idにKafkaのoffsetを組んで重複を潰す

Kafkaの取り込みは少なくとも1回の到達が前提です。リバランスやプロセス再起動をまたぐと、コミット前に処理していた区間が再送されます。ここでElasticsearch側に document_id を渡しておけば、同じイベントは同じIDで上書きされ、件数が二重に増える事態を避けられます。

output {
  elasticsearch {
    hosts => ["http://es-1:9200","http://es-2:9200"]
    index => "app-logs-%{+yyyy.MM.dd}"
    action => "index"
    document_id => "%{[@metadata][kafka][topic]}-%{[@metadata][kafka][partition]}-%{[@metadata][kafka][offset]}"
  }
}

トピック・パーティション・オフセットの三つ組はKafka内で一意なので、キーとして扱えます。ただしこの方式はデータストリームとは併用できません。data_stream は8.0以降の既定が auto で、有効なときのアクションは create になり、IDを指定する余地がなくなるためです。重複排除を優先するなら、通常のインデックスと index アクションを選ぶ判断になります。

Elasticsearchのインデックス分割と保持期間はILMへ寄せる

index の既定値は logstash-%{+yyyy.MM.dd} で、日付でインデックスが切り替わります。この既定のままだと、トピックが増えたときに全部が同じ系列へ混ざってしまう。decorate_events で付けたトピック名をインデックス名に差し込み、系統ごとに分けておくと、後から保持期間を変える判断がしやすくなります。

保持期間そのものはLogstash側で持たず、ILMの5フェーズ設計とmin_ageの起点に寄せます。ilm_enabled の既定は auto で、条件を満たせば自動で有効になる挙動です。収集経路とインデックス分割をまとめて見直す段階なら、ELKスタックの収集経路と保持期間の設計も併せて確認してください。

Logstashのバックプレッシャーと再送は詰まる場所を先に決める

連携が動き始めた後に問題になるのは、配送先が遅れたときに何が起きるかです。ここを設計しないまま本番へ出すと、Elasticsearchのわずかな遅延がKafkaの読み出し停止まで波及し、原因の切り分けに時間を取られます。

Logstashが詰まったときの逃げ場はKafkaのトピック側にある

Logstashのパイプラインは背圧を上流へ伝える作りになっています。Elasticsearchへの書き込みが遅れればフィルター段が詰まり、最終的にKafkaからの読み出し速度が落ちます。この挙動は障害ではなく、設計として期待される動きです。

Kafkaを前段に置く構成の利点はここに出ます。読み出しが遅れてもイベントはトピックに残り続けるため、保持期間の範囲内なら失われません。裏返せば、遅延が保持期間を食い潰すと取り返しがつかなくなる。ラグが保持期間に対してどれだけ余裕があるかを継続して見る必要があり、指標の選び方はKafka監視の指標選定とJMXの取得経路にまとめています。

再送の設計はat-least-onceを前提に重複ありで受ける

output側でKafkaへ書き込む場合、再試行の挙動も押さえておきます。2026年8月30日時点のドキュメントによれば retries は既定で成功するまで再試行し、プラグインの10.5.0以降は RetriableException と InterruptException のサブクラスだけが再試行の対象です。マッピング不正のような回復しない失敗を延々と再送し続ける挙動にはなりません。

output {
  kafka {
    bootstrap_servers => "kafka-1:9092,kafka-2:9092"
    topic_id => "app-logs"
    acks => "all"
    compression_type => "zstd"
    linger_ms => 100
    batch_size => 65536
  }
}

既定の batch_size は 16384 バイト、linger_ms は 5 ミリ秒、max_request_size は 1048576 バイトです。スループットを稼ぐなら滞留時間を延ばしてまとめて送る方向へ振りますが、その分だけ末端の遅延が増える。ログ配送であれば数十から数百ミリ秒の滞留は許容できることが多く、ここは遅延要件から決めます。

永続キューとDLQはKafkaと役割が重なる場所を先に見極める

Logstashには永続キューとデッドレターキューという二つの受け皿があります。前者は障害時のイベント保全、後者は処理できなかったイベントの隔離です。Kafkaを前段に置く構成では、前者の役割はトピックと重なります。

二重に持つとディスクを二か所で消費するだけになりやすい。input側にKafkaを置くなら永続キューは無効のままにし、保全はトピックの保持期間へ任せる割り切りが管理を軽くします。一方でデッドレターキューは役割が重ならないので、マッピング違反などで弾かれたイベントを拾うために残しておきます。

Kafka ConnectとLogstashは変換処理の重さで使い分ける

KafkaとElasticsearchをつなぐ手段はLogstashだけではありません。Kafka Connectのシンクコネクタでも同じことができます。どちらを選ぶかは好みではなく、あいだで何をするかで決まります。

変換が軽くて接続先が定番ならKafka Connectで足りる

コネクタは設定だけで動き、分散モードならワーカーを増やすだけで水平に伸ばせる仕組みです。変換がフィールドの付け外し程度で済み、接続先に公式コネクタが揃っているなら、Logstashを挟む理由は薄くなります。コネクタの構成要素とタスクの運用はKafka Connectの構成要素とコネクタ運用で扱っています。

運用面でも差が出ます。ConnectはタスクごとにREST経由で状態を確認でき、失敗したタスクの再起動もAPIで行えます。Logstashはプロセス単位の管理になるため、細かい制御を求めるとスクリプトを書く手間が増える。

grokや条件分岐が要る取り込みはLogstash側に寄せる

逆に、非構造のテキストをパースする、条件によって送り先を変える、外部データで補完するといった処理が入るなら、Logstashのほうが素直に書けます。Connectで同じことをやろうとすると、単一メッセージ変換の連結か、独自コネクタの実装に踏み込むことになる。

両方を使う構成も現実的です。DBからの取り込みはConnectのソースコネクタに任せ、Kafkaから先の整形と配送をLogstashが受け持つ形なら、それぞれの得意分野に寄せられます。実際、そう分けている現場のほうが管理は落ち着きます。

Logstashを挟む構成に踏み切る条件と見送る場面の判断軸を示す

ここまでの内容を、投資判断の形に落とします。Logstashを一段挟むということは、監視対象のプロセスとバージョン管理の対象が一つ増えるということでもあります。

Logstashを挟んでよいのは取り込み経路が増え続ける構成

踏み切ってよいのは、取り込み元が複数あって今後も増える見込みがあり、それぞれで整形処理が違う場合です。この条件なら、パイプライン定義を追加するだけで経路を足せる利点が、運用コストの増加を上回ります。

配送先がElasticsearch以外にも分かれる場合も同じです。条件分岐で複数のoutputへ配るのはLogstashが得意とする形で、Connectで同じ構成を組むよりも設定が読みやすくなります。取り込み元と配送先の掛け算が増えるほど、あいだに一段置く価値が高まる構造です。判断が難しい規模のときは、既存構成を前提にした設計相談としてデータ分析基盤構築・MLOps構築支援で受けています。

Logstashを見送ってよいのは単一経路で変換が要らない構成

見送ってよいのは、取り込み元が一つで、送り先も一つで、あいだの処理がほぼ無い場合です。この条件でLogstashを置くと、JVMを一つ増やして得られるものが実質ありません。BeatsからElasticsearchへ直接送る構成や、Connectのシンクコネクタで済ませる構成のほうが、監視すべき箇所が減ります。

検証段階のクラスタも同様です。まずは直結で通し、整形が要ることが分かった時点でLogstashを差し込む順序にすると、初期の切り分けが単純なまま保てる。あいだに一段入れるかどうかは、あとから足せる判断なので、先に入れておく必要はありません。

よくある質問

LogstashのKafkaプラグインは個別にインストールが必要ですか？

2026年8月30日時点では、inputとoutputの両方が logstash-integration-kafka という統合パッケージに同梱され、v12.1.0（2026年3月18日公開）が案内されていました。標準の配布物に含まれているため、通常は個別の導入操作なしで設定ファイルに書けます。バージョンを固定したい場合のみ、明示的な指定を検討してください。

consumer_threadsを増やしても取り込みが速くなりません

トピックのパーティション数がスレッド数を下回っている可能性が高いです。公式ドキュメントには、パーティション数を超えたスレッドは待機状態になり得る旨の記述があります。パーティション数を増やすか、Logstash側の並列度をパーティション数に合わせて下げ、ボトルネックが配送先にないかを先に切り分けてください。

Kafkaから同じログが二重にElasticsearchへ入ってしまいます

再送時の重複が素通りしています。decorate_events を basic 以上にしてKafka側のメタデータを載せ、Elasticsearch outputの document_id にトピック・パーティション・オフセットを組み合わせた値を渡してください。同じIDで上書きされるため件数が膨らみません。データストリームを使っている場合はIDを指定できないので、通常のインデックスへ切り替える判断が要ります。

enable_auto_commitはtrueとfalseのどちらにすべきですか？

欠落と重複のどちらを受け入れるかで決まります。既定の true は処理完了前にコミットが進む可能性があり、障害時に取りこぼしが出る。false にすると取りこぼしは減る代わりに重複が増えるため、前問の重複排除とセットで採用してください。監査ログのように欠落が許されない経路では後者を選びます。

fluentdやBeatsとの併用はできますか？

できます。BeatsやfluentdがKafkaへ書き込み、Logstashがそこから読み出す構成は一般的です。この形なら収集エージェントは軽いまま保て、整形処理はLogstash側に集約できます。ただし各段でイベント形式が変わるため、どの段でスキーマを確定させるかを先に決めておかないと、後段のマッピング調整が繰り返し発生します。

関連記事

• Apache Kafkaが選ばれる理由とは？特徴とメリットを徹底解説：連携先となるブローカーとトピックの構造を先に押さえられます。
• Logstashのgrokフィルターの使い方（構文とカスタムパターン）：Kafkaから受けた非構造ログを分解する具体的な書き方を確認できます。
• Kafka Connectとは？構成要素とコネクタ運用を実装視点で解説【2026年版】：Logstashと使い分ける際の対抗手段として比較できます。
• Elasticsearchのログ収集｜ELKスタックの収集経路とインデックス分割・保持期間の設計：収集経路全体の組み立て方を確認できます。
• Elasticsearchのデータ投入｜bulk APIのNDJSON形式と部分失敗の扱い、差分同期の設計：配送先で起きる部分失敗の扱いを詳しく確認できます。
• ElasticsearchのILM｜5フェーズの設計とmin_ageの起点・データストリームとの使い分け：保持期間の設計を配送側から切り離す方法が分かります。
• Kafka監視の設計｜ラグ・ISR・ディスクの指標選定とJMX取得経路を実装視点で解説：ラグと保持期間の余裕を継続して見る方法を確認できます。