Flume NG 学习笔记(十) Transaction、Sink、Source和Channel开发

发布时间:2020-07-27 22:42:46 作者:jackwxh
来源:网络 阅读:2187

目录(?)[+]

一、Transaction interface

Transaction接口是基于flume的稳定性考虑的。所有主要的组件(sources、sinks、channels)都必须使用Flume Transaction。我们也可以理解Transaction接口就是flume的事务,sources和sinks的发送数据与接受数据都是在一个Transaction里完成的。


从上图中可以看出,一个Transaction在Channel实现内实现。每一个连接到channel的source和sink都要获取一个Transaction对象。这Sources实际上使用了一个ChannelSelector接口来封装Transaction。存放事件到channel和从channel中提取事件的操作是在一个活跃的Transaction内执行的。

下面是官网例子


[java] view plain copy

  1. Channel ch = new MemoryChannel();  

  2. Transaction txn = ch.getTransaction();  

  3. txn.begin();  

  4. try {  

  5.   // This try clause includes whatever Channel operations you want to do  

  6.   

  7.   Event eventToStage = EventBuilder.withBody("Hello Flume!",  

  8.                        Charset.forName("UTF-8"));  

  9.   ch.put(eventToStage);  

  10.   // Event takenEvent = ch.take();  

  11.   // ...  

  12.   txn.commit();  

  13. catch (Throwable t) {  

  14.   txn.rollback();  

  15.   

  16.   // Log exception, handle individual exceptions as needed  

  17.   

  18.   // re-throw all Errors  

  19.   if (t instanceof Error) {  

  20.     throw (Error)t;  

  21.   }  

  22. finally {  

  23.   txn.close();  

  24. }  


上面的代码是一个很简单的Transaction示例,在自定义Source与自定义Sink中都要使用。

二、自定义Sink开发

Sink提取event数据从channel中,然后直接将数据发送到下一个flume agent中或者存储到外部库中。

Sink和channel的关联关系可以在配置文件中配置。有一个SinkRunner实例与每一个已配置的Sink关联,当Flume框架调用SinkRunner.start()方法时候,将创建一个新的线程来驱动这Sink。

这个线程将管理这个Sink的生命周期。Sink需要实现LifecycleAware接口的start()和stop()方法。start()方法用于初始化数据;stop()用于释放资源;process()是从channel中提取event数据和转发数据的核心方法。

这Sink需要实现Configurable接口以便操作配置文件。

下面是官网例子:

[java] view plain copy

  1. public class MySink extends AbstractSink implements Configurable {  

  2.   private String myProp;  

  3.   

  4.   @Override  

  5.   public void configure(Context context) {  

  6.     String myProp = context.getString("myProp""defaultValue");  

  7.   

  8.     // Process the myProp value (e.g. validation)  

  9.   

  10.     // Store myProp for later retrieval by process() method  

  11.     this.myProp = myProp;  

  12.   }  

  13.   

  14.   @Override  

  15.   public void start() {  

  16.     // Initialize the connection to the external repository (e.g. HDFS) that  

  17.     // this Sink will forward Events to ..  

  18.   }  

  19.   

  20.   @Override  

  21.   public void stop () {  

  22.     // Disconnect from the external respository and do any  

  23.     // additional cleanup (e.g. releasing resources or nulling-out  

  24.     // field values) ..  

  25.   }  

  26.   

  27.   @Override  

  28.   public Status process() throws EventDeliveryException {  

  29.     Status status = null;  

  30.   

  31.     // Start transaction  

  32.     Channel ch = getChannel();  

  33.     Transaction txn = ch.getTransaction();  

  34.     txn.begin();  

  35.     try {  

  36.       // This try clause includes whatever Channel operations you want to do  

  37.   

  38.       Event event = ch.take();  

  39.   

  40.       // Send the Event to the external repository.  

  41.       // storeSomeData(e);  

  42.   

  43.       txn.commit();  

  44.       status = Status.READY;  

  45.     } catch (Throwable t) {  

  46.       txn.rollback();  

  47.   

  48.       // Log exception, handle individual exceptions as needed  

  49.   

  50.       status = Status.BACKOFF;  

  51.   

  52.       // re-throw all Errors  

  53.       if (t instanceof Error) {  

  54.         throw (Error)t;  

  55.       }  

  56.     } finally {  

  57.       txn.close();  

  58.     }  

  59.     return status;  

  60.   }  

  61. }  

下面是测试例子:

[java] view plain copy

  1. import org.apache.flume.Channel;  

  2. import org.apache.flume.Context;  

  3. import org.apache.flume.Event;  

  4. import org.apache.flume.EventDeliveryException;  

  5. import org.apache.flume.Transaction;  

  6. import org.apache.flume.conf.Configurable;  

  7.   

  8. import org.apache.flume.sink.AbstractSink;  

  9.   

  10.   

  11. public class Custom_Sink extends AbstractSink implements Configurable {  

  12.       private String myProp;  

  13.      @Override  

  14.       public void configure(Context context) {  

  15.         String myProp = context.getString("myProp""defaultValue");  

  16.   

  17.         // Process the myProp value (e.g. validation)  

  18.   

  19.         // Store myProp for later retrieval by process() method  

  20.         this.myProp = myProp;  

  21.       }  

  22.   

  23.       @Override  

  24.       public void start() {  

  25.         // Initialize the connection to the external repository (e.g. HDFS) that  

  26.         // this Sink will forward Events to ..  

  27.       }  

  28.   

  29.       @Override  

  30.       public void stop () {  

  31.         // Disconnect from the external respository and do any  

  32.         // additional cleanup (e.g. releasing resources or nulling-out  

  33.         // field values) ..  

  34.       }  

  35.   

  36.       @Override  

  37.       public Status process() throws EventDeliveryException {  

  38.         Status status = null;  

  39.   

  40.         // Start transaction  

  41.         Channel ch = getChannel();  

  42.         Transaction txn = ch.getTransaction();  

  43.         txn.begin();  

  44.         try {  

  45.           // This try clause includes whatever Channel operations you want to do  

  46.             

  47.           Event event = ch.take();  

  48.           String out = new String(event.getBody());   

  49.           // Send the Event to the external repository.  

  50.           // storeSomeData(e);  

  51.           System.out.println(out);  

  52.             

  53.           txn.commit();  

  54.           status = Status.READY;  

  55.         } catch (Throwable t) {  

  56.           txn.rollback();  

  57.   

  58.           // Log exception, handle individual exceptions as needed  

  59.   

  60.           status = Status.BACKOFF;  

  61.   

  62.           // re-throw all Errors  

  63.           if (t instanceof Error) {  

  64.             throw (Error)t;  

  65.           }  

  66.         } finally {  

  67.           txn.close();  

  68.         }  

  69.         return status;  

  70.       }  

  71.   

  72. }  

上面的测试例子只输出事件的BODY信息,这里说明下直接用代码event.getBody().tostring() 输出是乱码。因为所有sink都是在Transaction里完成的,因此自定义开发sink是需要加上Transaction相关设置。

 

然后是测试配置,这里是自定义的jar 包是flumedev.Custom_Sink。注意,打包之后请放在目录$FLUME_HOME/lib下

[html] view plain copy

  1. #配置文件:custom_sink_case23.conf  

  2. # Name the components on this agent  

  3. a1.sources = r1  

  4. a1.sinks = k1  

  5. a1.channels = c1  

  6.   

  7. # Describe/configure the source  

  8. a1.sources.r1.type = syslogtcp  

  9. a1.sources.r1.port = 50000  

  10. a1.sources.r1.bind = 192.168.233.128  

  11. a1.sources.r1.channels = c1  

  12.   

  13. # Describe the sink  

  14. a1.sinks.k1.channel = c1  

  15. a1.sinks.k1.type = flumedev.Custom_Sink  

  16. #a1.sinks.k1.type =logger  

  17.   

  18. # Use a channel which buffers events in memory  

  19. a1.channels.c1.type = memory  

  20. a1.channels.c1.capacity = 1000  

  21. a1.channels.c1.transactionCapacity = 100  

#敲命令

flume-ng agent -cconf -f conf/custom_sink_case23.conf -n a1 -Dflume.root.logger=INFO,console

启动成功后

打开另一个终端输入,往侦听端口送数据

echo "testcustom_sink" | nc 192.168.233.128 50000

#在启动的终端查看console输出

Flume NG 学习笔记(十) Transaction、Sink、Source和Channel开发

可以看到数据正常输出。


三、自定义Source开发

Source从外面接收数据并把数据存入Channel中。很少有人用。

下面是官网的例子

[java] view plain copy

  1. public class MySource extends AbstractSource implements Configurable, PollableSource {  

  2.   private String myProp;  

  3.   

  4.   @Override  

  5.   public void configure(Context context) {  

  6.     String myProp = context.getString("myProp""defaultValue");  

  7.   

  8.     // Process the myProp value (e.g. validation, convert to another type, ...)  

  9.   

  10.     // Store myProp for later retrieval by process() method  

  11.     this.myProp = myProp;  

  12.   }  

  13.   

  14.   @Override  

  15.   public void start() {  

  16.     // Initialize the connection to the external client  

  17.   }  

  18.   

  19.   @Override  

  20.   public void stop () {  

  21.     // Disconnect from external client and do any additional cleanup  

  22.     // (e.g. releasing resources or nulling-out field values) ..  

  23.   }  

  24.   

  25.   @Override  

  26.   public Status process() throws EventDeliveryException {  

  27.     Status status = null;  

  28.   

  29.     // Start transaction  

  30.     Channel ch = getChannel();  

  31.     Transaction txn = ch.getTransaction();  

  32.     txn.begin();  

  33.     try {  

  34.       // This try clause includes whatever Channel operations you want to do  

  35.   

  36.       // Receive new data  

  37.       Event e = getSomeData();  

  38.   

  39.       // Store the Event into this Source's associated Channel(s)  

  40.       getChannelProcessor().processEvent(e)  

  41.   

  42.       txn.commit();  

  43.       status = Status.READY;  

  44.     } catch (Throwable t) {  

  45.       txn.rollback();  

  46.   

  47.       // Log exception, handle individual exceptions as needed  

  48.   

  49.       status = Status.BACKOFF;  

  50.   

  51.       // re-throw all Errors  

  52.       if (t instanceof Error) {  

  53.         throw (Error)t;  

  54.       }  

  55.     } finally {  

  56.       txn.close();  

  57.     }  

  58.     return status;  

  59.   }  

  60. }  


测试的话,主要针对Event e 这里进行传输数据,这里就不测试了。

 

四、自定义Channel开发

官网说待定。

下面是美团网的自定义Channel 开发,下面是链接

http://tech.meituan.com/mt-log-system-optimization.html

……

Flume本身提供了MemoryChannel和FileChannel。MemoryChannel处理速度快,但缓存大小有限,且没有持久化;FileChannel则刚好相反。我们希望利用两者的优势,在Sink处理速度够快,Channel没有缓存过多日志的时候,就使用MemoryChannel,当Sink处理速度跟不上,又需要Channel能够缓存下应用端发送过来的日志时,就使用FileChannel,由此我们开发了DualChannel,能够智能的在两个Channel之间切换。

其具体的逻辑如下:

[java] view plain copy

  1. /*** 

  2.  * putToMemChannel indicate put event to memChannel or fileChannel 

  3.  * takeFromMemChannel indicate take event from memChannel or fileChannel 

  4.  * */  

  5. private AtomicBoolean putToMemChannel = new AtomicBoolean(true);  

  6. private AtomicBoolean takeFromMemChannel = new AtomicBoolean(true);  

  7.   

  8. void doPut(Event event) {  

  9.         if (switchon && putToMemChannel.get()) {  

  10.               //往memChannel中写数据  

  11.               memTransaction.put(event);  

  12.   

  13.               if ( memChannel.isFull() || fileChannel.getQueueSize() > 100) {  

  14.                 putToMemChannel.set(false);  

  15.               }  

  16.         } else {  

  17.               //往fileChannel中写数据  

  18.               fileTransaction.put(event);  

  19.         }  

  20.   }  

  21.   

  22. Event doTake() {  

  23.     Event event = null;  

  24.     if ( takeFromMemChannel.get() ) {  

  25.         //从memChannel中取数据  

  26.         event = memTransaction.take();  

  27.         if (event == null) {  

  28.             takeFromMemChannel.set(false);  

  29.         }   

  30.     } else {  

  31.         //从fileChannel中取数据  

  32.         event = fileTransaction.take();  

  33.         if (event == null) {  

  34.             takeFromMemChannel.set(true);  

  35.   

  36.             putToMemChannel.set(true);  

  37.         }   

  38.     }  

  39.     return event;  

  40. }  

这里要说明下,官网是建议使用file channel,虽然它的效率比较低,但是它能保证数据完整性,而memory channel效率高,但是只能对数据丢失和重复不太敏感的业务使用


推荐阅读:
  1. Flume-1.6.0学习笔记(六)kafka source
  2. FLUME NG的基本架构

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

文章 开发 channel

上一篇:Planner .NET日历日程控件能给你的应用程序提供多种日历日程功能

下一篇:exchange server 2010安装CAS+HUB角色安装顺序

相关阅读

您好,登录后才能下订单哦!

密码登录
登录注册
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》