在发送到通道之前删除 null Flume 事件答案

【问题标题】：Drop null Flume events before sending to channel在发送到通道之前删除 null Flume 事件
【发布时间】：2014-01-16 14:28:56
【问题描述】：

我在一本 Flume 的书中读到，如果在拦截器的拦截方法中，如果一个事件返回为 null，则该事件将被丢弃。因此，我创建了一个自定义拦截器，它根据条件将事件返回为 null，例如：

public Event intercept(Event event) {
    // TODO Auto-generated method stub
    Event finalEvent = event;
    check = new String(event.getBody(),Charsets.UTF_8);

    if(check.matches("([0-9]-.+?-.+?-[0-9][0-9]+)")){

        try {
            fileWriter.append(new String(event.getBody(),Charsets.UTF_8)+ "\n");
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        finalEvent = null;
    }
    System.out.println("Event is : " + finalEvent);
    return finalEvent;
}

拦截器发出 null 事件，但文件通道仍将其作为空传递给 HDFS 接收器。为什么事件没有被丢弃？我使用 Spooling 目录作为源。

【问题讨论】：

也许你可以做一个测试？让您的代码始终返回 null 并查看是否所有事件都已删除。如果是，那么问题可能出在匹配逻辑而不是水槽配置中。
@AndrewEhrlich 我也有类似的问题。就我而言，我已经根据正则表达式修改了拦截器中的事件正文。在接收器（即 HDFS）中，我可以看到与根据应用逻辑修改的正则表达式匹配的行，但是应该删除的其他事件未修改，因此我的正则表达式工作正常。我在代码。我还需要做点别的吗？
你能给你的水槽配置链接吗？也许那里有什么配置错误。

标签： hadoop null interceptor cloudera flume

【解决方案1】：

让我们看看会发生什么。你使用 Spooling 目录作为源，源调用函数 processEventBatch(events)，在 processEventBatch() 内部：

events = interceptorChain.intercept(events);//use your custom interceptor
...
eventQueue.add(event); // add user event to queue,even the event == null

如果 Spooling 目录源使用 processEvent()，那么你的拦截器将起作用：在 processEvent() 内部：

event = interceptorChain.intercept(event);
if (event == null) {
  //null event then return !! intercept works!!
  return;
}

所以你应该修改 processEventBatch()，然后做：

if (event == null){
    //dont add to eventQueue
}

【讨论】：

我在拦截器类中使用了您的解决方案，而不是自定义源，因为我已经在使用拦截器进行其他检查。谢谢！！

【解决方案2】：

在我的拦截器类中，方法 intercept(Event event) 包含有关数据如何流动的逻辑（如问题中所写），如果不满足正则表达式，则返回 null，在方法 intercept(List<Event> events) 中为 null 事件排除了哪些服务器的目的。以下是intercept(List<Event> events) 代码：

public List<Event> intercept(List<Event> events) 
    {
          List<Event> interceptedEvents = new ArrayList<Event>(events.size());
          for (Event event : events) 
          {
              // Intercept any event
              Event interceptedEvent = intercept(event);
              if(interceptedEvent!=null)
                  interceptedEvents.add(interceptedEvent);
          }

          return interceptedEvents;
    }

【讨论】：