使用flume反序列化Json文件并沉入HDFS

1 回答

• 1

  最好的猜测是编写一个自定义拦截器，将您的JSON转换为所需的HDFS格式 . 它还具有填充可在hdfs路径中使用的标头的好处 .

  以下是配置拦截器的方法：

  agent_slave_1.sources.source1_1.interceptors = my_intercptor
  agent_slave_1.sources.source1_1.interceptors.my_intercptor.type = com.mycompany.MyInteceptor
  

  这个类看起来像这样：

  public class MyInteceptor implements Interceptor, Interceptor.Builder {
  
      private MyInteceptor interceptor;
  
      @Override
      public void initialize() {
  
  
      }
  
      @Override
      public Event intercept(Event event) {
          String bjson = event.getBody()));
          // decode your json, e.g. Jackson
          MyDecodedJsonObject record; // pseudo class
          event.getHeaders().put("timestamp", record.getTimestamp().toString());
          String newBody = record.getA() + "\t" + record.getB();
          event.setBody(newBody.getBytes())
          return event;
      }
  
      @Override
      public List<Event> intercept(List<Event> events) {
  
          for (Iterator<Event> iterator = events.iterator(); iterator.hasNext();) {
              Event next = intercept(iterator.next());
              if (next == null) {
                  iterator.remove();
              }
          }
          return events;
      }
  
      @Override
      public void close() {
  
  
      }
  
      @Override
      public Interceptor build() {
          return interceptor;
      }
  
      @Override
      public void configure(Context context) {
  
          interceptor = new MyInteceptor();
      }
  
  }
  

  不要忘记将这个类打包在jar中并将其放入flume的lib目录中 .

  回复于 2024-04-20T12:17:49+08:00