Flume HDFS Sink使用及源码分析

cike0415

HDFS Sink介绍
　　Flume导入数据HDFS，目前只支持创建序列化（sequence）文件和文本（text）文件。还支持这两个文件的压缩。文件可以根据运行的时间，数据的大小和时间的数量来进行周期性的滚动（关闭当前文件产生新的文件）。也可以根据数据属性分区，例如根据时间戳或机器分区。HDFS目录路径可以包含格式化的转义字符，生成目录路径可以通过格式化转移字符（escape sequences），HDFS sink通过这些转义字符生成一个目录或者文件去存储Event。当然在Flume中使用HDFS Sink的话，需要添加HDFS相关的Jar，这样Flume就能使用Hadoop的jar和Hadoop集群交互。注：Hadoop必须支持sync()。


以下是HDFS Sink支持的转义字符：


名称  描述

%{host}


替代Event Header被命名为“host”的值，支持任意的Header name。




%t


Unix毫秒时间




%a


短的周名称，例如：Mon, Tue, ...




%A


周名称全称，例如：Monday, Tuesday, ...




%b


短的月名称，例如：(Jan, Feb, ...




%B


月名称全称，例如：January, February, ...




%c


日期和时间，例如：Thu Mar 3 23:05:25 2005




%d


每个月的某一天，例如：01 - 31




%e


每个月的某一天（没有填充0）例如：1,2,3,4---31




%D


日期；像：%m/%d/%y




%H


小时(00..23)




%I


小时(01..12)




%j


每个年的某一天，例如：001..366




%k


小时，例如：0..23




%m


月份，例如：01..12




%n


月份，例如：1..12




%M


分钟，例如：00..59




%p


am 或 pm




%s


从1970-01-01 00:00:00 UTC到现在的毫秒数




%S


秒，例如：00..59




%y


两位数的年份，例如：00..99




%Y


年份，例如：2010




%z


+hhmm 数字时区，例如：-0400


文件在使用的时候以".tmp"为后缀，一旦文件关闭，扩展名将被移除。

注：跟时间相关的转移序列，Key为“timestamp”必须存在在Event的Headers中（除非hdfs.useLocalTimeStamp设置为true）


NameDefaultDescription

channel


–









type


–


组件的名称，必须为：HDFS




hdfs.path


–


HDFS目录路径，例如：hdfs://namenode/flume/webdata/




hdfs.filePrefix


FlumeData


HDFS目录中，由Flume创建的文件前缀。




hdfs.fileSuffix


–


追加到文件的后缀，例如：.txt




hdfs.inUsePrefix


–


文件正在写入时的前缀。




hdfs.inUseSuffix


.tmp


文件正在写入时的后缀。




hdfs.rollInterval


30


当前写入的文件滚动间隔，默认30秒生成一个新的文件 (0 = 不滚动)




hdfs.rollSize


1024


以文件大小触发文件滚动，单位字节(0
= 不滚动)




hdfs.rollCount


10



以写入的事件数触发文件滚动。(0 = 不滚动)




hdfs.idleTimeout


0


超时多久以后关闭无效的文件。(0 = 禁用自动关闭的空闲文件)但是还是可能因为网络等多种原因导致，正在写的文件始终没有关闭，从而产生tmp文件




hdfs.batchSize


100


有多少Event后，写到文件才刷新到HDFS。




hdfs.codeC


–


压缩编解码器，可以使用：gzip, bzip2, lzo, lzop, snappy




hdfs.fileType


SequenceFile


文件格式：通常使用SequenceFile（默认）,DataStream或者CompressedStream
(1)DataStream不能压缩输出文件，请不用设置hdfs.codeC编码解码器。

(2)CompressedStream要求设置hdfs.codeC来制定一个有效的编码解码器。




hdfs.maxOpenFiles


5000


HDFS中允许打开文件的数据，如果数量超过了，最老的文件将被关闭。




hdfs.callTimeout


10000


允许HDFS操作的毫秒数，例如：open，write, flush, close。如果很多HFDS操作超时，这个配置应该增大。




hdfs.threadsPoolSize


10



每个HDFS sink的HDFS的IO操作线程数（例如：open，write）







hdfs.rollTimerPoolSize


1


每个HDFS sink调度定时文件滚动的线程数。




hdfs.kerberosPrincipal


–


安全访问HDFS Kerberos的主用户。




hdfs.kerberosKeytab


–


安全访问HDFSKerberos keytab




hdfs.proxyUser














hdfs.round


false


时间戳应该被四舍五入。(如果为true，会影响所有的时间，除了t%)




hdfs.roundValue


1


四舍五入的最高倍数（单位配置在hdfs.roundUnit），但是要小于当前时间。




hdfs.roundUnit


second


四舍五入的单位，包含：second,minuteorhour.




hdfs.timeZone


Local Time


时区的名称，主要用来解决目录路径。例如：America/Los_Angeles




hdfs.useLocalTimeStamp


false


使用本地时间替换转义字符。 (而不是event header的时间戳)




hdfs.closeTries


0


在发起一个关闭命令后，HDFS sink必须尝试重命名文件的次数。如果设置为1，重命名失败后，HDFS sink不会再次尝试重命名该文件，这个文件处于打开状态，并且用.tmp作为扩展名。如果为0，Sink会一直尝试重命名，直至重命名成功。如果文件 失败，这个文件可能一直保持打开状态，但是这种情况下数据是完整的。文件将会在Flume下次重启时被关闭。




hdfs.retryInterval


180


在几秒钟之间连续尝试关闭文件。每个关闭请求都会有多个RPC往返Namenode，因此设置的太低可能导致Namenode超负荷，如果设置0或者更小，如果第一次尝试失败的话，该Sink将不会尝试关闭文件。并且把文件打开，或者用“.tmp”作为扩展名。




serializer


TEXT


可能的选项包括avro_event或继承了EventSerializer.Builder接口的类名。




serializer.*
















关于round：

a1.sinks.k1.hdfs.round=true
a1.sinks.k1.hdfs.roundValue=10
a1.sinks.k1.hdfs.roundUnit=minute
上面的配置将四舍五入配置到10分钟，例如：一个事件的时间戳是

11:54:34 AM, June 12, 2012 将导致hdfs的路径变为：/flume/events/2012-06-12/1150/00

源码分析
　　configure(Context context)：主要用于加载配置文件。


public void configure(Context context) {
this.context = context;
//HDFS目录路径，例如：hdfs://namenode/flume/webdata/,也可以用/flume/webdata/，这样要把Hadoop的配置文件放到classpath
filePath = Preconditions.checkNotNull(
context.getString("hdfs.path"), "hdfs.path is required");
//HDFS目录中，由Flume创建的文件前缀。
fileName = context.getString("hdfs.filePrefix", defaultFileName);
//文件后缀
this.suffix = context.getString("hdfs.fileSuffix", defaultSuffix);
//文件正在写入时的前缀。
inUsePrefix = context.getString("hdfs.inUsePrefix", defaultInUsePrefix);//文件正在写入时的后缀。
inUseSuffix = context.getString("hdfs.inUseSuffix", defaultInUseSuffix);
//时区的名称，主要用来解决目录路径。例如：America/Los_Angeles
String tzName = context.getString("hdfs.timeZone");
timeZone = tzName == null ? null : TimeZone.getTimeZone(tzName);
rollInterval = context.getLong("hdfs.rollInterval", defaultRollInterval);//当前写入的文件滚动间隔，默认30秒生成一个新的文件 (0 = 不滚动)
rollSize = context.getLong("hdfs.rollSize", defaultRollSize);//以文件大小触发文件滚动，单位字节(0 = 不滚动)
rollCount = context.getLong("hdfs.rollCount", defaultRollCount);
//有多少Event后，写到文件才刷新到HDFS。
batchSize = context.getLong("hdfs.batchSize", defaultBatchSize);
//超时多久以后关闭无效的文件。(0 = 禁用自动关闭的空闲文件)但是还是可能因为网络等多种原因导致，正在写的文件始终没有关闭，从而产生tmp文件
idleTimeout = context.getInteger("hdfs.idleTimeout", 0);
//压缩编解码器，可以使用：gzip, bzip2, lzo, lzop, snappy
String codecName = context.getString("hdfs.codeC");
//文件格式：通常使用SequenceFile（默认）, DataStream 或者 CompressedStrea
//(1)DataStream不能压缩输出文件，请不用设置hdfs.codeC编码解码器。
//(2)CompressedStream要求设置hdfs.codeC来制定一个有效的编码解码器。
fileType = context.getString("hdfs.fileType", defaultFileType);
//HDFS中允许打开文件的数据，如果数量超过了，最老的文件将被关闭。
maxOpenFiles = context.getInteger("hdfs.maxOpenFiles", defaultMaxOpenFiles);
//允许HDFS操作的毫秒数，例如：open，write, flush, close。如果很多HFDS操作超时，这个配置应该增大。
callTimeout = context.getLong("hdfs.callTimeout", defaultCallTimeout);
//允许HDFS操作的毫秒数，例如：open，write, flush, close。如果很多HFDS操作超时，这个配置应该增大。
//每个HDFS sink的HDFS的IO操作线程数（例如：open，write）
threadsPoolSize = context.getInteger("hdfs.threadsPoolSize", defaultThreadPoolSize);
//每个HDFS sink调度定时文件滚动的线程数。
rollTimerPoolSize = context.getInteger("hdfs.rollTimerPoolSize", defaultRollTimerPoolSize);
//每个HDFS sink调度定时文件滚动的线程数。
String kerbConfPrincipal = context.getString("hdfs.kerberosPrincipal");
//安全认证
String kerbKeytab = context.getString("hdfs.kerberosKeytab");
String proxyUser = context.getString("hdfs.proxyUser");
tryCount = context.getInteger("hdfs.closeTries", defaultTryCount);
if(tryCount <= 0) {
LOG.warn("Retry count value : " + tryCount + " is not " +
"valid. The sink will try to close the file until the file " +
"is eventually closed.");
tryCount = defaultTryCount;
}
retryInterval = context.getLong("hdfs.retryInterval",
defaultRetryInterval);
if(retryInterval <= 0) {
LOG.warn("Retry Interval value: " + retryInterval + " is not " +
"valid. If the first close of a file fails, " +
"it may remain open and will not be renamed.");
tryCount = 1;
}
Preconditions.checkArgument(batchSize > 0,
"batchSize must be greater than 0");
if (codecName == null) {
codeC = null;
compType = CompressionType.NONE;
} else {
codeC = getCodec(codecName);
// TODO : set proper compression type
compType = CompressionType.BLOCK;
}
// Do not allow user to set fileType DataStream with codeC together
// To prevent output file with compress extension (like .snappy)
if(fileType.equalsIgnoreCase(HDFSWriterFactory.DataStreamType)
&& codecName != null) {
throw new IllegalArgumentException("fileType: " + fileType +
" which does NOT support compressed output. Please don't set codeC" +
" or change the fileType if compressed output is desired.");
}
if(fileType.equalsIgnoreCase(HDFSWriterFactory.CompStreamType)) {
Preconditions.checkNotNull(codeC, "It's essential to set compress codec"
+ " when fileType is: " + fileType);
}
// get the appropriate executor
this.privExecutor = FlumeAuthenticationUtil.getAuthenticator(
kerbConfPrincipal, kerbKeytab).proxyAs(proxyUser);

//时间戳应该被四舍五入。(如果为true，会影响所有的时间，除了t%)
needRounding = context.getBoolean("hdfs.round", false);
if(needRounding) {
//四舍五入的单位
String unit = context.getString("hdfs.roundUnit", "second");
if (unit.equalsIgnoreCase("hour")) {
this.roundUnit = Calendar.HOUR_OF_DAY;
} else if (unit.equalsIgnoreCase("minute")) {
this.roundUnit = Calendar.MINUTE;
} else if (unit.equalsIgnoreCase("second")){
this.roundUnit = Calendar.SECOND;
} else {
LOG.warn("Rounding unit is not valid, please set one of" +
"minute, hour, or second. Rounding will be disabled");
needRounding = false;
}
//四舍五入的最高倍数
this.roundValue = context.getInteger("hdfs.roundValue", 1);
if(roundUnit == Calendar.SECOND || roundUnit == Calendar.MINUTE){
Preconditions.checkArgument(roundValue > 0 && roundValue <= 60,
"Round value" +
"must be > 0 and <= 60");
} else if (roundUnit == Calendar.HOUR_OF_DAY){
Preconditions.checkArgument(roundValue > 0 && roundValue <= 24,
"Round value" +
"must be > 0 and <= 24");
}
}
this.useLocalTime = context.getBoolean("hdfs.useLocalTimeStamp", false);
if(useLocalTime) {
clock = new SystemClock();
}
if (sinkCounter == null) {
//<span style="color:#000000;">计数器</span>
sinkCounter = new SinkCounter(getName());
}
}
































































































































































































































































按照Flume的生命周期，先启动start方法：
  @Override
public void start() {
String timeoutName = "hdfs-" + getName() + "-call-runner-%d";
//线程池用于event写入HDFS文件
callTimeoutPool = Executors.newFixedThreadPool(threadsPoolSize,
new ThreadFactoryBuilder().setNameFormat(timeoutName).build());
String rollerName = "hdfs-" + getName() + "-roll-timer-%d";
//该线程池用来滚动文件
timedRollerPool = Executors.newScheduledThreadPool(rollTimerPoolSize,
new ThreadFactoryBuilder().setNameFormat(rollerName).build());
//该LinkedHashMap用来存储文件的绝对路径以及对应的BucketWriter
this.sfWriters = new WriterLinkedHashMap(maxOpenFiles);
sinkCounter.start();
super.start();
}所有的Event，经Source后发送的Channel，再由Channel传入到Sink，主要调用Sink的process方法实现事务：
public Status process() throws EventDeliveryException {
Channel channel = getChannel();//获取Channel
Transaction transaction = channel.getTransaction();//获取事务
List<BucketWriter> writers = Lists.newArrayList();//初始化BucketWriter列表，BucketWriter是操作HDFS主类。
transaction.begin();
try {
int txnEventCount = 0;
for (txnEventCount = 0; txnEventCount < batchSize; txnEventCount++) {//批量处理
Event event = channel.take();//获取Event
if (event == null) {
break;
}
// reconstruct the path name by substituting place holders
String realPath = BucketPath.escapeString(filePath, event.getHeaders(),
timeZone, needRounding, roundUnit, roundValue, useLocalTime);//格式化HDFS路径，根据转义字符
String realName = BucketPath.escapeString(fileName, event.getHeaders(),
timeZone, needRounding, roundUnit, roundValue, useLocalTime);//格式化文件名称，根据转义字符
//写入HDFS的绝对路径
String lookupPath = realPath + DIRECTORY_DELIMITER + realName;
BucketWriter bucketWriter;
HDFSWriter hdfsWriter = null;
// Callback to remove the reference to the bucket writer from the
// sfWriters map so that all buffers used by the HDFS file
// handles are garbage collected.
WriterCallback closeCallback = new WriterCallback() {
@Override
public void run(String bucketPath) {
LOG.info("Writer callback called.");
synchronized (sfWritersLock) {
sfWriters.remove(bucketPath);
}
}
};
synchronized (sfWritersLock) {
//根据HDFS的绝对路径获取对应的BucketWriter对象
bucketWriter = sfWriters.get(lookupPath);
// we haven't seen this file yet, so open it and cache the handle
if (bucketWriter == null) {
//初始化BuchetWriter对象
hdfsWriter = writerFactory.getWriter(fileType);
bucketWriter = initializeBucketWriter(realPath, realName,
lookupPath, hdfsWriter, closeCallback);
//放入Map
sfWriters.put(lookupPath, bucketWriter);
}
}
// track the buckets getting written in this transaction
if (!writers.contains(bucketWriter)) {
//如果BucketWriter列表没有正在写的文件——bucketWriter，则加入
writers.add(bucketWriter);
}
// Write the data to HDFS
try {
//将event写入bucketWriter对应的文件中
bucketWriter.append(event);
} catch (BucketClosedException ex) {
LOG.info("Bucket was closed while trying to append, " +
"reinitializing bucket and writing event.");
hdfsWriter = writerFactory.getWriter(fileType);
bucketWriter = initializeBucketWriter(realPath, realName,
lookupPath, hdfsWriter, closeCallback);
synchronized (sfWritersLock) {
sfWriters.put(lookupPath, bucketWriter);
}
bucketWriter.append(event);
}
}
if (txnEventCount == 0) {
//这次事务没有处理任何event
sinkCounter.incrementBatchEmptyCount();
} else if (txnEventCount == batchSize) {
//一次处理batchSize个event
sinkCounter.incrementBatchCompleteCount();
} else {
//channel中剩余的events不足batchSize
sinkCounter.incrementBatchUnderflowCount();
}
// flush all pending buckets before committing the transaction
//获取List里面的BucketWriter的所有数据都刷新到HDFS
for (BucketWriter bucketWriter : writers) {
//如果使用转义字符生成文件名或路径，可能还没有满足其他滚动生成新文件的条件，就有新文件产生，
//在这种情况下，例如为hdfs.idleTimeout=0，那么就可能会在HDFS中出现很多.tmp后缀的文件。因为调用flush没有关闭该文件。
bucketWriter.flush();
}
//提交事务
transaction.commit();
if (txnEventCount < 1) {
return Status.BACKOFF;
} else {
sinkCounter.addToEventDrainSuccessCount(txnEventCount);
return Status.READY;
}
} catch (IOException eIO) {
transaction.rollback();//事务回滚
LOG.warn("HDFS IO error", eIO);
return Status.BACKOFF;
} catch (Throwable th) {
transaction.rollback();
LOG.error("process failed", th);
if (th instanceof Error) {
throw (Error) th;
} else {
throw new EventDeliveryException(th);
}
} finally {
transaction.close();//关闭事务
}
}


HDFS Sink流程分析：　　1，通过configure(Context context)和start()方法初始化Sink
　　2，SinkRunner的线程调用process()方法，循环处理批量的Event，如果Event为null，就跳出循环。
　　3，有Event数据，先格式化HDFS的文件路径和文件名，即：realPath和realName。realPath+realName就是完整HDFS路径：lookupPath，然后根据lookupPath获取BucketWriter对象。
　　4，BucketWriter对象不存在，则先构建根据fileType构建一个HDFSWriter 对象。然后初始化BucketWriter对象。最后将对象放到sfWriters中，表示正在写的文件。


  public HDFSWriter getWriter(String fileType) throws IOException {
if (fileType.equalsIgnoreCase(SequenceFileType)) {
//通过SequenceFile.Writer写入文件
return new HDFSSequenceFile();
} else if (fileType.equalsIgnoreCase(DataStreamType)) {
//通过FSDataOutputStream
return new HDFSDataStream();
} else if (fileType.equalsIgnoreCase(CompStreamType)) {
return new HDFSCompressedDataStream();
} else {
throw new IOException("File type " + fileType + " not supported");
}
}
　　HDFSSequenceFile：configure(context)方法会首先获取写入格式writeFormat即参数"hdfs.writeFormat"，org.apache.flume.sink.hdfs.SequenceFileSerializerType定义了一下三个：

  Writable(HDFSWritableSerializer.Builder.class),//默认的
Text(HDFSTextSerializer.Builder.class),
Other(null);


再获取是否使用HDFS本地文件系统"hdfs.useRawLocalFileSystem"，默认是flase不使用；然后获取writeFormat的所有配置信息serializerContext；然后根据writeFormat和serializerContext构造SequenceFileSerializer的对象serializer。
　　HDFSDataStream：configure(context)方法先获取serializerType类型，默认是TEXT(BodyTextEventSerializer.Builder.class)，其他的还包含：

public enum EventSerializerType {
TEXT(BodyTextEventSerializer.Builder.class),
HEADER_AND_TEXT(HeaderAndBodyTextEventSerializer.Builder.class),
AVRO_EVENT(FlumeEventAvroEventSerializer.Builder.class),
OTHER(null);


再获取是否使用HDFS本地文件系统"hdfs.useRawLocalFileSystem"，默认是flase不使用；最后获取serializer的所有配置信息serializerContext。serializer的实例化在HDFSDataStream.doOpen(Configuration conf, Path dstPath, FileSystem hdfs)方法中实现的。　　HDFSCompressedDataStream：configure和HDFSDataStream.configure(context)类似，serializerType的类型也一样。serializer的实例化是在HDFSCompressedDataStream.open(String filePath, CompressionCodec codec, CompressionType cType)方法中实现。
　　5，bucketWriter实例化后存放到sfWriters中，并且判断是否在writers变量的List中，如果不存在，就放入List，这样后面就可以对bucketWriter统一flush了。
　　6，bucketWriter.append(event);

  public synchronized void append(final Event event)
throws IOException, InterruptedException {
checkAndThrowInterruptedException();//检查当前线程是否被中断
// If idleFuture is not null, cancel it before we move forward to avoid a
// close call in the middle of the append.
if(idleFuture != null) {
idleFuture.cancel(false);
// There is still a small race condition - if the idleFuture is already
// running, interrupting it can cause HDFS close operation to throw -
// so we cannot interrupt it while running. If the future could not be
// cancelled, it is already running - wait for it to finish before
// attempting to write.
if(!idleFuture.isDone()) {
try {
idleFuture.get(callTimeout, TimeUnit.MILLISECONDS);
} catch (TimeoutException ex) {
LOG.warn("Timeout while trying to cancel closing of idle file. Idle" +
" file close may have failed", ex);
} catch (Exception ex) {
LOG.warn("Error while trying to cancel closing of idle file. ", ex);
}
}
idleFuture = null;
}
// If the bucket writer was closed due to roll timeout or idle timeout,
// force a new bucket writer to be created. Roll count and roll size will
// just reuse this one
if (!isOpen) {
if (closed) {
throw new BucketClosedException("This bucket writer was closed and " +
"this handle is thus no longer valid");
}
open();//一个文件已经完成将isOpen设置为false，则新建一个文件
}
// check if it's time to rotate the file
if (shouldRotate()) {//检查文件的行数及大小，判断是否要关闭文件后重新生成文件。
boolean doRotate = true;
if (isUnderReplicated) {
if (maxConsecUnderReplRotations > 0 &&
consecutiveUnderReplRotateCount >= maxConsecUnderReplRotations) {
doRotate = false;
if (consecutiveUnderReplRotateCount == maxConsecUnderReplRotations) {
LOG.error("Hit max consecutive under-replication rotations ({}); " +
"will not continue rolling files under this path due to " +
"under-replication", maxConsecUnderReplRotations);
}
} else {
LOG.warn("Block Under-replication detected. Rotating file.");
}
consecutiveUnderReplRotateCount++;
} else {
consecutiveUnderReplRotateCount = 0;
}
if (doRotate) {
close();
open();//新建一个文件
}
}
// write the event
try {
sinkCounter.incrementEventDrainAttemptCount();
callWithTimeout(new CallRunner<Void>() {
@Override
public Void call() throws Exception {
writer.append(event); // could block 往HDFS写入数据。
return null;
}
});
} catch (IOException e) {
LOG.warn("Caught IOException writing to HDFSWriter ({}). Closing file (" +
bucketPath + ") and rethrowing exception.",
e.getMessage());
try {
close(true);
} catch (IOException e2) {
LOG.warn("Caught IOException while closing file (" +
bucketPath + "). Exception follows.", e2);
}
throw e;
}
// update statistics
processSize += event.getBody().length;
eventCounter++;
batchCounter++;
if (batchCounter == batchSize) {
flush();
}
}


打开新文件分为两类：　　第一类不需要压缩


  public void open(String filePath) throws IOException {
open(filePath, null, CompressionType.NONE);
}


第二类要压缩
  public void open(String filePath, CompressionCodec codeC,
CompressionType compType) throws IOException {
Configuration conf = new Configuration();
Path dstPath = new Path(filePath);
FileSystem hdfs = dstPath.getFileSystem(conf);
open(dstPath, codeC, compType, conf, hdfs);
}注：HDFSDataStream是不支持压缩的，所以直接调用第一类的open方法。　　在open方法中，如果按时间滚动的rollInterval不为0，则创建Callable，放入timedRollFuture中rollInterval秒之后关闭文件，默认是30s写一个文件。
　　最后writer.append(event)是真正写数据到HDFS，writer分如下三种情况：

HDFSSequenceFile：append(event)方法，会先通过serializer.serialize(e)把event处理成一个Key和一个Value。
　　serializer为HDFSWritableSerializer：
　　Key：


private Object getKey(Event e) {
String timestamp = e.getHeaders().get("timestamp");//获取header的timesteamp
long eventStamp;
if (timestamp == null) {//timestamp不存在就拿系统的当前时间
eventStamp = System.currentTimeMillis();
} else {
eventStamp = Long.valueOf(timestamp);
}
return new LongWritable(eventStamp);//将时间封装成LongWritable
}Value：
  private BytesWritable makeByteWritable(Event e) {
BytesWritable bytesObject = new BytesWritable();
bytesObject.set(e.getBody(), 0, e.getBody().length);
return bytesObject;
}serializer为HDFSTextSerializer：　　Key同上，Value：

  private Text makeText(Event e) {
Text textObject = new Text();
textObject.set(e.getBody(), 0, e.getBody().length);
return textObject;
}


writer为HDFSDataStream：　　直接调用serializer.write(e)，serializer分三种：
　　org.apache.flume.serialization.BodyTextEventSerializer直接读取body写入OutputStream流中，然后在最后加"\n"。
　　org.apache.flume.serialization.HeaderAndBodyTextEventSerializer将e.getHeaders() + " " +e.getBody()写入数据流，然后根据配置看是否要加"\n"
　　org.apache.flume.serialization.AvroEventSerializer将event整体写入dataFileWriter。
　　

　　然后appned方法更新统计，processSize统计文件大小；eventCounter统计文件行数；batchCounter是统计最近一次flush之后的处理的event数；

如果处理的event数量达到batchSize的大小，则刷新到HDFS，flush()方法会首先执行writer.sync()即写入HDFS，然后将batchCounter置为0，根据fileType的不同writer也会有很多写入类型：


HDFSSequenceFile：sync()方法执行SequenceFile.Writer.syncFs()将数据写入HDFS中；

HDFSDataStream：sync()方法执行

HDFSCompressedDataStream：sync()方法先执行serializer.flush()：只有FlumeEventAvroEventSerializer的flush()方法也有实现dataFileWriter.flush()，其他俩BodyTextEventSerializer和HeaderAndBodyTextEventSerializer均未实现flush()方法。然后执行outStream.flush()和outStream.sync()将数据刷新至HDFS中。
　　

　　7，回到HDFSEventSink.process()方法中，会根据这次事务处理的event数量更新相应的统计；
　　8，遍历writers，挨个刷新BucketWriter至HDFS；
　　9，最后提交事务，异常回滚，关闭事务。
　　最后停止：

  @Override
public void stop() {
// do not constrain close() calls with a timeout
synchronized (sfWritersLock) {//获取对象锁
//遍历对象锁
for (Entry<String, BucketWriter> entry : sfWriters.entrySet()) {
LOG.info("Closing {}", entry.getKey());
//关闭BucketWriter，flush到HDFS
try {
entry.getValue().close();
} catch (Exception ex) {
LOG.warn("Exception while closing " + entry.getKey() + ". " +
"Exception follows.", ex);
if (ex instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
}
}
}
// shut down all our thread pools
ExecutorService toShutdown[] = {callTimeoutPool, timedRollerPool};
for (ExecutorService execService : toShutdown) {
execService.shutdown();
try {
while (execService.isTerminated() == false) {
execService.awaitTermination(
Math.max(defaultCallTimeout, callTimeout), TimeUnit.MILLISECONDS);
}
} catch (InterruptedException ex) {
LOG.warn("shutdown interrupted on " + execService, ex);
}
}
callTimeoutPool = null;
timedRollerPool = null;
synchronized (sfWritersLock) {
sfWriters.clear();
sfWriters = null;
}
sinkCounter.stop();
super.stop();
}