FlumeでSourceとカスタムSinkをカスタマイズします.

カスタムソース
Sourceは、Flume Agentへのデータの受信を担当するコンポーネントです.Sourceコンポーネントは、avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslop、http、legacyを含む様々なタイプ、様々なフォーマットのログデータを処理することができます.公式に提供されているsourceのタイプはすでに多いですが、実際に開発されたニーズに満足できない場合があります.公式説明によると、カスタムMySourceはAbstractSourceクラスを引き継ぎ、ConfigrableとPollable Sourceインターフェースを実現する必要があります.

  ：  flume    ，          ，
          。    flume       。
      maven module，    
    <dependencies>
    <dependency>
        <groupId>org.apache.flume</groupId>
        <artifactId>flume-ng-core</artifactId>
        <version>1.9.0</version>
</dependency>

</dependencies>  

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.source.AbstractSource;

/**
 *  for         
 *                  body       
 */

public class Mysource extends AbstractSource implements Configurable, PollableSource {
     
    private String prefis;
    private String suffix;

    //           
    @Override
    public Status process() throws EventDeliveryException {
     
        Status status = null;
        try {
     
            for (int i = 0; i < 5; i++) {
     
                SimpleEvent event = new SimpleEvent();
                event.setBody((prefis + "===" + i + "====" + suffix).getBytes());
                getChannelProcessor().processEvent(event);
                Thread.sleep(2000);
            }
            status=Status.READY;
        } catch (Exception e) {
     
            e.printStackTrace();
            status = Status.BACKOFF;
        }
        return status;
    }

    @Override
    public long getBackOffSleepIncrement() {
     
        return 0;
    }

    @Override
    public long getMaxBackOffSleepInterval() {
     
        return 0;
    }

    @Override
    public void configure(Context context) {
     
        prefis = context.getString("prefix");
        suffix = context.getString("suffix", "biedata");
    }

テスト:作成したコードを包装し、flume/lib/下に置いてください.プロファイルの変更:flume/jobでvim Mysource-flume-logger.comに以下のような設定ファイルを追加します.

# Name the components on this agent
a2.sources = r1 
a2.sinks = k1
a2.channels = c1 

# Describe/configure the source
a2.sources.r1.type =  MySource     
a2.sources.r1.prefix = qianzhui
a2.sources.r1.suffix = houzhui

# Describe the sink
a2.sinks.k1.type = logger


# Use a channel which buffers events in memory
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1 

試験結果を観察します.2 sごとにデータをコンソールに印刷します.
カスタムSink
Sinkは、Channel内のイベントを絶えずポーリングし、それらを一括的に除去し、これらのイベントをメモリまたはインデックスシステムに書き込み、または別のFlume Agentに送信する.Sinkは完全に事務的です.Channelからデータを大量に削除する前に、各SinkはChannelで一つの事務を開始します.一括イベントが成功すると、ストレージシステムまたは次のFlume Agentに書き出され、SinkはChanelを利用してトランザクションを提出する.一旦事務が提出されると、Channelは自分の内部バッファからイベントを削除します.公式説明によると、カスタムMySinkはAbstractSinkクラスを引き継ぎ、Configrableインターフェースを実現する必要がある.

  ：  flume    ，  Sink             ，
      。     flume         。

import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *   logger4j       
 *                      
 */
public class MySink extends AbstractSink implements Configurable {
     
    //       
    private String prefix;
    private String suffix;
    //     logger  
    private Logger logger = LoggerFactory.getLogger(MySink.class);
    @Override
    //sink        
    public Status process() throws EventDeliveryException {
     

        //1.    
        Status status=null;
        //2.   sink   channel
        Channel channel = getChannel();
        //3. channel    
        Transaction transaction = channel.getTransaction();
        //4.    
        transaction.begin();
        try {
     
            //5. channnel   
            Event event = channel.take();
            //6.       event    null
            if (event!=null){
     
                //7.  logger    (            )
                logger.info( prefix+"-->"+new String(event.getBody())+"+suffix);
            }
            //8.     
            transaction.commit();
            //9.         
            status=Status.READY;
        }catch (Exception e){
     
            e.printStackTrace();
            //10.        
            status=Status.BACKOFF;
            //11.     
            transaction.rollback();
        }finally {
     
            //12.    
            transaction.close();
        }
        return status;
    }

    @Override
    public void configure(Context context) {
     
    prefix=context.getString("aaa");
    suffix=context.getString("suffix", "abc");
    }
}

テスト:作成したコードを包装し、flume/lib/下に置いてください.プロファイルを追加:

# Name the components on this agent
a2.sources = r1 
a2.sinks = k1
a2.channels = c1 

# Describe/configure the source
a2.sources.r1.type = netcat
a2.sources.r1.bind = localhost
a2.sources.r1.port = 44444

# Describe the sink
a2.sinks.k1.type = MySink     
a2.sinks.k1.aaa = qianzhui


# Use a channel which buffers events in memory
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1 

結果を観察すればいいです.