flume LineDeserializer Line length exceeds max (2048), truncating line!扩大一行数据量大小的采集上限

简介

    在一次使用flume+kafka+sparkstreaming架构处理日志时，出现一个很奇怪的问题：日志中的某一行数据总会被切分成了多行，总的输出日志行数也比原始日志文件多出了几十行，导致具体的处理逻辑中出现各种错误。经过排查，定位到问题是出现在flume上。 在flume中采集的一行数据量是有大小限制的，默认为2048，即2KB，而我的日志中有些行的大小已经超过了这个限制，然后flume就会将之切分成多行。同时flume也会在采集的时候弹出警告信息：

[WARN - org.apache.flume.serialization.LineDeserializer.readLine(LineDeserializer.java:143)] Line length exceeds max (2048), truncating line!

下面是flume1.7的源码中出现问题的地方（本段代码，位于flume源码包org\apache\flume\serialization类中）。

 public static final String MAXLINE_KEY = "maxLineLength";
  public static final int MAXLINE_DFLT = 2048; //默认的大小

  LineDeserializer(Context context, ResettableInputStream in) {
    this.in = in;
    this.outputCharset = Charset.forName(
        context.getString(OUT_CHARSET_KEY, CHARSET_DFLT));
    this.maxLineLength = context.getInteger(MAXLINE_KEY, MAXLINE_DFLT);
    this.isOpen = true;
  }

/********/ 

private String readLine() throws IOException {
    StringBuilder sb = new StringBuilder();
    int c;
    int readChars = 0;
    while ((c = in.readChar()) != -1) {
      readChars++;

      // FIXME: support \r\n
      if (c == '\n') {
        break;
      }

      sb.append((char)c);

      if (readChars >= maxLineLength) {//这里就会进行大小判断
        logger.warn("Line length exceeds max ({}), truncating line!",
            maxLineLength);
        break;
      }
    }

    if (readChars > 0) {
      return sb.toString();
    } else {
      return null;
    }
  }

修改及测试

          因此，接下来就可以修改源码，将上面代码中的2048，改成更大的值，我改成4096，暂时够用了。然后编译即可。我为了方便，只编译org\apache\flume\serialization这个类。也可以用maven等去编译，同时注意编译使用的JDK要和flume运行环境的JDK大版本一致。然后放入flume的core核心包里替换原有的class(可以先备份core包)

要修改的jar包:



修改值：



替换class:



接下来重新采集文件，警告信息消失，sparkstreaming中消费的数据也正常了。