このトピックでは、E-MapReduce-Flume が監査ログを HDFS に移動する例を、Flume の使用方法を説明する例として取り上げます。

背景

E-MapReduce は、バージョン 3.19.0 以降の EMR-Flume の クラスター管理をサポートしています。 クラスター管理を使用すると、Web ページから Flume Agent の設定および管理を行えます。 この例では、マスターインスタンスで Flume Agent を起動して、ローカルディスクに保存されている監査ログを収集し、Avro プロトコルを使用してログをコアインスタンスに送信します。 Failover Sink Processor は、マスターインスタンスからデータを受信するようにコアインスタンスで設定および開始されます。 次に、シンクを使用してデータを HDFS に移動します。 Flume を使用するその他のシナリオについては、「Flume の使用方法」をご参照ください。

準備

EMR クラスターを作成して、オプションのサービスから Flume を選択します。 詳細については、「クラスターの作成」をご参照ください。

手順

  • コアインスタンスで Flume Agent を設定して起動します。
    たとえば、次の図に示すように、Flume Agent を emr-worker-1 ノードのコアインスタンスグループ向けに設定します。
    1. [サービス構成] セクションで、構成を次のように設定します。
      default-agent.sinks.default-sink.type hdfs
      default-agent.channels.default-channel.type file
      default-agent.sources.default-source.type avro
      deploy_node_hostname emr-worker-1
    2. [カスタム設定] をクリックして、次の設定を追加します。
      default-agent.sinks.default-sink.hdfs.path 高可用性クラスターの場合、アドレスには hdfs://emr-cluster/path format が使用されます。
      default-agent.sinks.default-sink.hdfs.fileType DataStream
      default-agent.sinks.default-sink.hdfs.rollSize 0
      default-agent.sinks.default-sink.hdfs.rollCount 0
      default-agent.sinks.default-sink.hdfs.rollInterval 86400
      default-agent.sinks.default-sink.hdfs.batchSize 51200
      default-agent.sources.default-source.bind 0.0.0.0
      default-agent.sources.default-source.port 必要に応じて、値を設定します。
      default-agent.channels.default-channel.transactionCapacity 51200
      default-agent.channels.default-channel.dataDirs チャンネルがイベントデータを格納するパス。
      Default-agent.channels.default-channel.checkpointDir チェックポイントファイルが格納されるパス。
      default-agent.channels.default-channel.capacity hdfs ロールに基づいて値を設定します。
    3. 設定を保存して、Flume Agent を起動します。
    4. [履歴] をクリックします。 [ステータス] 列に [成功] と表示されたら、 [コンポーネントのデプロイメント] タブページの [ステータス] 列に emr-worker-1 ノードの Flume エージェントが STARTED と表示されます。emr-worker-1 ノードで Flume エージェントが起動したら、他のワーカーノードでも Flume エージェントを起動します。 たとえば、emr-worker-2 ノードで Flume エージェントを起動するには、次の設定項目を変更します。
      deploy_node_hostname ノードのホスト名。
      default-agent.sinks.default-sink.hdfs.path 高可用性クラスターの場合、アドレスには hdfs://emr-cluster/path 形式が使用されます。
    5. 設定を保存し、すべてのコンポーネントを起動して、ターゲットノードとして emr-worker-2 を選択します。
  • マスターインスタンスで Flume エージェントを設定して起動します。

    例:

    次のようにエージェントを設定します。
    additional_sinks k1
    deploy_node_hostname emr-header-1
    default-agent.sources.default-source.type taildir
    default-agent.sinks.default-sink.type avro
    default-agent.channels.default-channel.type file
    The reader configurations are as follows:
    設定項目 値
    default-agent.sources.default-source.filegroups f1
    default-agent.sources.default-source.filegroups.f1 /mnt/disk1/log/hadoop-hdfs/hdfs-audit.log. *
    default-agent.sources.default-source.positionFile 位置ファイルが保存されているパス。
    default-agent.channels.default-channel.checkpointDir チェックポイントファイルが保存されているパス。
    default-agent.channels.default-channel.dataDirs チャネルがイベントデータを格納するパス。
    default-agent.channels.default-channel.capacity 必要に応じて値を設定します。
    default-agent.sources.default-source.batchSize 2000
    default-agent.channels.default-channel.transactionCapacity 2000
    default-agent.sources.default-source.ignoreRenameWhenMultiMatching true
    default-agent.sinkgroups g1
    default-agent.sinkgroups.g1.sinks default-sink k1
    default-agent.sinkgroups.g1.processor.type failover
    default-agent.sinkgroups.g1.processor.priority.default-sink 10
    default-agent.sinkgroups.g1.processor.priority.k1 5
    default-agent.sinks.default-sink.hostname emr-worker-1 ノードの IP アドレス。
    default-agent.sinks.default-sink.port emr-worker-1 ノードの Flume Agent のポート。
    default-agent.sinks.k1.hostname emr-worker- ノードの IP アドレス。
    default-agent.sinks.k1.port emr-worker-2 ノードの Flume Agent のポート。
    default-agent.sinks.default-sink.batch-size 2000
    default-agent.sinks.k1.batch-size 2000
    default-agent.sinks.k1.type avro
    default-agent.sinks.k1.channel default-channel

同期結果の表示

HDFS コマンドを使用すると、FlumeData.${timestamp} という名前のファイルにデータが書き込まれていることがわかります。"timestamp" はファイルがいつ作成されたかを示します。