Flume-データストリームを管理します。 パート1

1. Flumeとは何ですか？

2.重要な用語

• イベント（イベント） -追加のメタ情報を持つデータの単位。 イベントの構造は、POST要求に似ています。
  
  • 見出し（ヘッダー） -メタ情報、「キー」-「値」のペアのセット。
  • コンテンツ（本体） -実際には、すべてが開始されるためのデータ。 バイト[]として渡されます。
  
  
• クライアントは、Flumeホストに関してデータを提供する外部サービスです。
  
  
• ソース -データの受信を担当します。 同時に、FlumeはEventDrivenSourceとPollableSourceの2種類のソースを提供します。 最初のケースでは、ソースはイベントをチャネルに追加するタイミングを決定します（たとえば、HTTPSourceはHTTP要求の受信時にイベントを追加します）。 EventDrivenSourceとは対照的にPollableSourceは受動的です-Flumeは、定期的に新しいイベントのソースをポーリングします。
  
  
• シンクは、データを次の処理段階に転送するコンポーネントです。 別のFlumeノード、ファイルシステム、HDFSなどを使用できます。
  
  
• チャネルは、データを転送するときにバッファとして機能するコンポーネントです。 チャネルは受動的なコンポーネントであり、それ自体ではアクションを開始しません。 ソースはイベントをチャネルに追加し、ドレインは空にします。
  
  
• エージェントは、Flumeコンポーネント（ソース、チャネル、ドレイン）が機能するプロセスです。 一般に、JVMインスタンス。 1つのノードに複数のエージェントを含めることができます。

3. Flumeノードの構造

• ノードには、多くのソース、チャネル、およびドレインを組み込むことができます。
• 1つのソースが複数のチャネルにイベントを追加できます（ルールに従って複製または配布します）。
• 複数のソースが1つのチャネルでイベントをスタックします。
• 1つのドレインは、1つのチャネルのみで機能します。
• 複数のシンクが1つのチャネルからイベントを受け取ることができます（均等に、または何らかのバランス規則に従って）。

4. Flumeノードの構成と起動

• Memchannel（RAMを使用してイベントを保存するチャネル）。
• NetCatソース。
• ロガーシンク（イベントをコンソールに出力するストック）。

### ====================   ==================== ### #    ,      : ,    # <agent>.sources -  ,   (    : my_source) my-agent.sources = my-source # <agent>.channels -     my-agent.channels = my-channel # <agent>.sinks -      my-agent.sinks = my-sink ### ====================  my_source ================== ### #   - netcat (    Flume   -, #          ,  ..,  ) my-agent.sources.my-source.type = netcat # ,     my-agent.sources.my-source.bind = 0.0.0.0 my-agent.sources.my-source.port = 11111 #    (  ,  ),     my-agent.sources.my-source.channels = my-channel ### ====================  my_channel ================== ### #      Flume - memory (   ,     ),      my-agent.channels.my-channel.type = memory #  , _  my-agent.channels.my-channel.capacity = 10000 #      (  ,    "") my-agent.channels.my-channel.transactionCapacity = 100 ### ====================  my_sink ================== ### #   - ,     (      ) my-agent.sinks.my-sink.type = logger #       my-agent.sinks.my-sink.channel = my-channel #      logger -         my-agent.sinks.my-sink.maxBytesToLog = 256 

1. Hadoopクラスターでは、Cloudera Managerを使用します（ この記事では、これを行う方法について詳しく説明しています）。
2. Flumeライブラリを使用するJavaサービスとして。

  <repositories> <repository> <id>cloudera</id> <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url> </repository> </repositories> <dependencies> <dependency> <groupId>org.apache.flume</groupId> <artifactId>flume-ng-sdk</artifactId> <version>1.5.0-cdh5.3.0</version> </dependency> <dependency> <groupId>org.apache.flume</groupId> <artifactId>flume-ng-node</artifactId> <version>1.5.0-cdh5.3.0</version> </dependency> </dependencies> 

 package ru.flume.samples; import org.apache.flume.node.Application; public class FlumeLauncher { public static void main(String[] args) { //    Log4j       System.setProperty("log4j.configuration", "file:/flume/config/log4j.properties"); //  Flume  : Application.main(new String[]{ "-f", "/flume/config/sample.conf", //      "-n", "my-agent" //   }); } } 

  INFO main conf.FlumeConfiguration-処理：my-sink
 INFO main conf.FlumeConfiguration-追加されたシンク：my-sinkエージェント：my-agent
 INFO main conf.FlumeConfiguration-処理：my-sink
 INFO main conf.FlumeConfiguration-処理：my-sink
 INFO main conf.FlumeConfiguration-検証後のflume設定には、エージェントの設定が含まれます：[my-agent]
 INFOメインnode.AbstractConfigurationProvider-チャネルの作成
 INFO main channel.DefaultChannelFactory-チャンネルmy-channelタイプメモリのインスタンスの作成
 INFOメインnode.AbstractConfigurationProvider-作成されたチャネルmy-channel
 INFO main source.DefaultSourceFactory-ソースmy-sourceのインスタンスの作成、netcatと入力
 INFO main sink.DefaultSinkFactory-シンクのインスタンスの作成：my-sink、タイプ：logger
 INFOメインnode.AbstractConfigurationProvider-[my-source、my-sink]に接続されたチャネルmy-channel
 INFOメインnode.Application-新しい構成の開始：
 {
     sourceRunners：{
         my-source = EventDrivenSourceRunner：{ 
            ソース：org.apache.flume.source.NetcatSource {
                名前：my-source、
                状態：アイドル
             }
         }
     } 
     sinkRunners：{
         my-sink = SinkRunner：{ 
            ポリシー：org.apache.flume.sink.DefaultSinkProcessor@77f03bb1 counterGroup：{name：null counters：{}} 
         }
     }            
    チャンネル：{
         my-channel = org.apache.flume.channel.MemoryChannel {
            名前：my-channel
         }
     }
 }
 INFOメインノードアプリケーション-開始チャネルmy-channel
 INFOメインnode.Application-チャネルを待機しています：my-channelが開始します。  500ミリ秒スリープ
 INFO lifecycleSupervisor-1-0 instrumentation.MonitoredCounterGroup-タイプ：CHANNEL、名前：my-channel：の監視対象カウンターグループ：新規MBeanが正常に登録されました。
 INFO lifecycleSupervisor-1-0 instrumentation.MonitoredCounterGroup-コンポーネントタイプ：CHANNEL、名前：my-channel started
 INFOメインノードアプリケーション-Sink my-sinkの開始
 INFOメインnode.Application-開始ソースmy-source
 INFO lifecycleSupervisor-1-1 source.NetcatSource-ソース開始
 INFO lifecycleSupervisor-1-1 source.NetcatSource-作成されたserverSocket：sun.nio.ch.ServerSocketChannelImpl [/ 0：0：0：0：0：0：0：0：11111] 

 Message 1 Message 2 Message 3 

 nc 127.0.0.1 11111 < test.txt 

 sink.LoggerSink - Event: { headers:{} body: 4D 65 73 73 61 67 65 20 31 0D Message 1. } sink.LoggerSink - Event: { headers:{} body: 4D 65 73 73 61 67 65 20 32 0D Message 2. } sink.LoggerSink - Event: { headers:{} body: 4D 65 73 73 61 67 65 20 33 0D Message 3. } 

5. Flumeノードチェーン

 node1.sources = my-source node1.channels = my-channel #   2 : node1.sinks = my-sink1 my-sink2 node1.sources.my-source.type = netcat node1.sources.my-source.bind = 0.0.0.0 node1.sources.my-source.port = 11111 node1.sources.my-source.channels = my-channel node1.channels.my-channel.type = memory node1.channels.my-channel.capacity = 10000 node1.channels.my-channel.transactionCapacity = 100 #      avro,        #      ,       #      node1.sinks.my-sink1.type = avro node1.sinks.my-sink1.channel = my-channel node1.sinks.my-sink1.hostname = 127.0.0.1 node1.sinks.my-sink1.port = 11112 node1.sinks.my-sink1.batch-size = 100 node1.sinks.my-sink2.type = avro node1.sinks.my-sink2.channel = my-channel node1.sinks.my-sink2.hostname = 127.0.0.1 node1.sinks.my-sink2.port = 11113 node1.sinks.my-sink2.batch-size = 100 

 node2.sources = my-source node2.channels = my-channel node2.sinks = my-sink #    1    avro,      avro node2.sources.my-source.type = avro node2.sources.my-source.bind = 0.0.0.0 node2.sources.my-source.port = 11112 node2.sources.my-source.channels = my-channel node2.channels.my-channel.type = memory node2.channels.my-channel.capacity = 10000 node2.channels.my-channel.transactionCapacity = 100 node2.sinks.my-sink.type = logger node2.sinks.my-sink.channel = my-channel node2.sinks.my-sink.maxBytesToLog = 256 

 node3.sources = my-source node3.channels = my-channel node3.sinks = my-sink #    1    avro,      avro node3.sources.my-source.type = avro node3.sources.my-source.bind = 0.0.0.0 node3.sources.my-source.port = 11113 node3.sources.my-source.channels = my-channel node3.channels.my-channel.type = memory node3.channels.my-channel.capacity = 10000 node3.channels.my-channel.transactionCapacity = 100 node3.sinks.my-sink.type = logger node3.sinks.my-sink.channel = my-channel node3.sinks.my-sink.maxBytesToLog = 256 

 Application.main(new String[]{"-f", "/flume/config/node1.conf", "-n", "node1"}); //     : //Application.main(new String[]{"-f", "/flume/config/node2.conf", "-n", "node2"}); //Application.main(new String[]{"-f", "/flume/config/node3.conf", "-n", "node3"}); 

おわりに

• ヘッダーとチャネルセレクター（チャネルセレクター）の使用。
• Flumeの「戦闘」コンポーネント：
  • アブロソース
  • ファイルチャンネル
  • アブロシンク;
  • HDFSシンク;
  • ファイルロールシンク。
• Flumeノードステータスの監視。

使用されたソースと便利なリンク

• Apache Flume公式ページ
• 公式Flumeコンポーネントチューニングガイド
• Hadoopパート2：Flumeを介したデータの収集| Selectelのブログは、ClouderaでFlumeをセットアップすることに関する記事です。
• Hari Shreedharan：Flumeの使用は、 Flumeの機能を説明する良い本です。