Simple Log Service:Flume を使用したログデータの取得

仕組み

Apache Flume は、Source-Channel-Sink のデータフローモデルを採用しています。aliyun-log-flume プラグインは、SLS を Flume パイプラインに接続するためのカスタム Sink およびカスタム Source を提供します。

• Sink： Flume Channel からデータを受け取り、SLS Logstore に書き込みます。Flume を介して他のシステムから SLS にデータを取り込む場合にこのコンポーネントを使用します。

• Source： SLS Logstore からログデータを取得し、Flume Channel に送信します。Flume を介して SLS のログデータを他のシステムに送信する場合にこのコンポーネントを使用します。

Channel は Source と Sink の間のバッファーとして機能します。Flume には Memory Channel や File Channel など、組み込みのチャネルタイプが用意されています。詳細については、「Apache Flume User Guide」をご参照ください。

プラグインのソースコードおよびリリースノートについては、「GitHub 上の aliyun-log-flume」をご参照ください。

前提条件

作業を開始する前に、以下の要件を満たしていることを確認してください。

• Java： JDK 1.8 以降がインストールされていること。

• Apache Flume： Flume 1.8.0 以降がインストールされていること。Flume のダウンロード方法については、「Apache Flume ダウンロードページ」をご参照ください。

• SLS リソース： SLS プロジェクトおよび Logstore が作成済みであること。詳細については、SLS ドキュメントをご参照ください。

• AccessKey ペア： AccessKey ID および AccessKey Secret を取得済みであること。セキュリティの観点から、RAM ユーザーの AccessKey ペアを使用することを推奨します。詳細については、「AccessKey ペア」をご参照ください。

プラグインのインストール

1. Flume をダウンロードしてインストールします。詳細については、「Apache Flume ダウンロードページ」をご参照ください。

2. aliyun-log-flume プラグインの JAR ファイルをダウンロードし、<FLUME_HOME>/lib ディレクトリに保存します。ダウンロードリンク：aliyun-log-flume-1.9.jar

3. <FLUME_HOME>/conf ディレクトリに flumejob.conf という名前の構成ファイルを作成します。Sink の構成については「Sink の構成」をご参照ください。Source の構成については「Source の構成」をご参照ください。

4. Flume を起動します。

      bin/flume-ng agent -n agent -c conf -f conf/flumejob.conf

Sink の構成

SLS Sink を使用して、他のシステムから Flume を介して SLS Logstore にデータを書き込みます。Sink は、Flume イベントを SLS ログエントリに変換するための次の 3 つのシリアル化モードをサポートしています。

# コンポーネントに名前を付けます
agent.sources = avroSrc
agent.channels = memCh
agent.sinks = slsSink

# Avro Source を構成します
agent.sources.avroSrc.type = avro
agent.sources.avroSrc.bind = 0.0.0.0
agent.sources.avroSrc.port = 4141
agent.sources.avroSrc.channels = memCh

# Memory Channel を構成します
agent.channels.memCh.type = memory
agent.channels.memCh.capacity = 1000
agent.channels.memCh.transactionCapacity = 100

# SLS Sink を構成します
agent.sinks.slsSink.type = com.aliyun.loghub.flume.sink.LoghubSink
agent.sinks.slsSink.channel = memCh
agent.sinks.slsSink.endpoint = http://cn-hangzhou.log.aliyuncs.com
agent.sinks.slsSink.project = your-project
agent.sinks.slsSink.logstore = your-logstore
agent.sinks.slsSink.accessKeyId = your-access-key-id
agent.sinks.slsSink.accessKey = your-access-key-secret
agent.sinks.slsSink.batchSize = 1000
agent.sinks.slsSink.serializer = DELIMITED
agent.sinks.slsSink.columns = col1,col2,col3
agent.sinks.slsSink.separatorChar = ,

Source の構成

SLS Source を使用して、SLS Logstore からログデータを取得し、Flume を介してダウンストリームシステムに配信します。Source は、SLS ログエントリを Flume イベントに変換するための次の 2 つの逆シリアル化モードをサポートしています。

# コンポーネントに名前を付けます
agent.sources = slsSrc
agent.channels = memCh
agent.sinks = loggerSink

# SLS Source を構成します
agent.sources.slsSrc.type = com.aliyun.loghub.flume.source.LoghubSource
agent.sources.slsSrc.channels = memCh
agent.sources.slsSrc.endpoint = http://cn-hangzhou.log.aliyuncs.com
agent.sources.slsSrc.project = your-project
agent.sources.slsSrc.logstore = your-logstore
agent.sources.slsSrc.accessKeyId = your-access-key-id
agent.sources.slsSrc.accessKey = your-access-key-secret
agent.sources.slsSrc.deserializer = JSON
agent.sources.slsSrc.sourceAsField = true
agent.sources.slsSrc.tagAsField = true
agent.sources.slsSrc.timeAsField = true
agent.sources.slsSrc.consumerGroup = flume-consumer
agent.sources.slsSrc.initialPosition = begin

# Memory Channel を構成します
agent.channels.memCh.type = memory
agent.channels.memCh.capacity = 1000
agent.channels.memCh.transactionCapacity = 100

# Logger Sink を構成します
agent.sinks.loggerSink.type = logger
agent.sinks.loggerSink.channel = memCh

Channel の構成

Channel は、Flume パイプラインにおいて Source と Sink を接続するバッファーです。aliyun-log-flume プラグインは、標準の Flume Channel タイプすべてで動作します。最も一般的なオプションは次の 2 つです。

• Memory Channel： イベントをメモリに格納します。高スループットを実現しますが、Flume エージェントプロセスが再起動するとイベントが失われます。データ損失が許容されるユースケースに適しています。

    agent.channels.memCh.type = memory
    agent.channels.memCh.capacity = 10000
    agent.channels.memCh.transactionCapacity = 1000

• File Channel： イベントをディスクに永続化します。スループットは低下しますが、耐久性を確保できます。データ損失が許容されない本番ワークロードに適しています。

    agent.channels.fileCh.type = file
    agent.channels.fileCh.checkpointDir = /var/flume/checkpoint
    agent.channels.fileCh.dataDirs = /var/flume/data
    agent.channels.fileCh.capacity = 1000000
    agent.channels.fileCh.transactionCapacity = 10000

重要： Channel の transactionCapacity は、Sink または Source で設定された batchSize 以上の値に設定してください。トランザクション容量がバッチサイズより小さい場合、Sink または Source は 1 つのトランザクション内でバッチを完了できず、エラーが発生します。

参考情報

• GitHub 上の aliyun-log-flume プラグイン

• Apache Flume User Guide

• SLS エンドポイント

• AccessKey ペアを取得する