经过测试发现,当source端单event的body数据大于16字节后,输出到目标只剩下16字节。进过多源代码的分析,发现,源代码中进行了截取。
在LoggerSink.java中:
if (event != null) {
if (logger.isInfoEnabled()) {
logger.info("Event: " + EventHelper.dumpEvent(event));
}
}
我们去看EventHelper.java的dumpEvent方法:
private static final int DEFAULT_MAX_BYTES = 16;
public static String dumpEvent(Event event) {
return dumpEvent(event, DEFAULT_MAX_BYTES);
}
public static String dumpEvent(Event event, int maxBytes) {
StringBuilder buffer = new StringBuilder();
if (event == null || event.getBody() == null) {
buffer.append("null");
} else if (event.getBody().length == 0) {
// do nothing... in this case, HexDump.dump() will throw an exception
} else {
byte[] body = event.getBody();
byte[] data = Arrays.copyOf(body, Math.min(body.length, maxBytes));
ByteArrayOutputStream out = new ByteArrayOutputStream();
try {
HexDump.dump(data, 0, out, 0);
String hexDump = new String(out.toByteArray());
// remove offset since it's not relevant for such a small dataset
if(hexDump.startsWith(HEXDUMP_OFFSET)) {
hexDump = hexDump.substring(HEXDUMP_OFFSET.length());
}
buffer.append(hexDump);
} catch (Exception e) {
if(LOGGER.isInfoEnabled()) {
LOGGER.info("Exception while dumping event", e);
}
buffer.append("...Exception while dumping: ").append(e.getMessage());
}
String result = buffer.toString();
if(result.endsWith(EOL) && buffer.length() > EOL.length()) {
buffer.delete(buffer.length() - EOL.length(), buffer.length()).toString();
}
}
return "{ headers:" + event.getHeaders() + " body:" + buffer + " }";
}
不难看出,在event处理过程中,发生了数据截取操作。
分享到:
相关推荐
Flume-ng-1.6.0-cdh.zip 内压缩了 3 个项目,分别为:flume-ng-1.6.0-cdh5.5.0.tar.gz、flume-ng-1.6.0-cdh5.7.0.tar.gz 和 flume-ng-1.6.0-cdh5.10.1.tar.gz,选择你需要的版本。
flume-ng安装
flume-ng-sql-source-1.5.2.jar从数据库中增量读取数据到hdfs中的jar包
flume-ng-sql-source-release-1.5.2.jar 用flume-ng-sql-source 从数据库抽取数据到kafka,支持sql
flume-ng-sql-source-1.5.3.jar,flume采集mysql数据jar包,将此文件拖入FLUME_HOME/lib目录下,如果是CM下CDH版本的flume,则放到/opt/cloudera/parcels/CDH-xxxx/lib/flume-ng/lib下,同样需要的包还有mysql-...
flume-ng-1.6.0-cdh5.5.0.tar.gz
flume-ng-sql-source-1.5.2源码
flume-ng-1.5.0-cdh5.3.6.rarflume-ng-1.5.0-cdh5.3.6.rar flume-ng-1.5.0-cdh5.3.6.rar flume-ng-1.5.0-cdh5.3.6.rar flume-ng-1.5.0-cdh5.3.6.rar flume-ng-1.5.0-cdh5.3.6.rar flume-ng-1.5.0-cdh5.3.6.rar flume...
flume-ng-sql-source-1.5.1 flume连接数据库 很好用的工具
flume-ng-1.6.0 cdh5.7.0安装包,稳定版本。大家可以自由下载
包含flume-ng-sql-source-1.5.1&flume;-ng-sql-source-1.4.1 此内容均为网上下载
flume-ng-sql-source-1.4.3.jar
Flume-ng在windows环境搭建并测试+log4j日志通过Flume输出到HDFS 11111
flume1.9采集数据入存入elasticsearch6.2.4,flume1.9本身只支持低版本的elasticsearch,基于apache-flume-1.9.0-src的flume-ng-sinks/flume-ng-elasticsearch-sink源码修改,支持es6.2.4,打的包,直接替换flume/...
flume-ng-sql-source实现oracle增量数据读取 flume连接oracle增量数据读取
Flume是Cloudera提供的一个高可用的,高可靠的,分布式的海量日志采集、聚合和传输的系统,Flume支持在日志系统中定制各类数据发送方,用于收集数据;同时,Flume提供对数据进行简单处理,并写到各种数据接受方(可...
Flume是Cloudera提供的一个高可用的,高可靠的,分布式的海量日志采集、聚合和传输的系统,Flume支持在日志系统中定制各类数据发送方,用于收集数据;同时,Flume提供对数据进行简单处理,并写到各种数据接受方(可...
flume-ng-sdk-1.6.0.ja,如果想要通过log4j将日志直接导入到flume需要导入这个jar包