上一篇说了HDFSEventSink的实现，这里根据hdfs sink的配置和调用分析来看下sink中整个hdfs数据写入的过程：
线上hdfs sink的几个重要设置
hdfs.path = hdfs://xxxxx/%{logtypename}/%Y%m%d/%H：
hdfs.rollInterval = 60
hdfs.rollSize = 0 //想让文件只根据实际来roll
hdfs.rollCount = 0
hdfs.batchSize = 2000
hdfs.txnEventMax = 2000
hdfs.fileType = DataStream
hdfs.writeFormat = Text这里说下和类相关的hdfs.fileType和hdfs.writeFormat，一个定义了文件流式用的类，一个定义了具体的数据序列化的类.
1)hdfs.fileType 有3个可选项：SequenceFile/DataStream/CompressedStream，DataStream可以想象成hdfs的textfile,默认是SequenceFileType，CompressedStream是用于压缩时设置
2)hdfs.writeFormat 定义了3种序列化方法,TEXT只写Event的body部分，HEADER_AND_TEXT写Event的body和header,AVRO_EVENT是avro的序列化方式
上面的设置，其数据写入流程大概如下：
SinkRunner.process->SinkProcessor.process->HDFSEventSink.process->HDFSEventSink.append->BucketWriter.append->HDFSWriter.append->HDFSDataStream.append->BodyTextEventSerializer.write->java.io.OutputStream.write
简单说下：
在HDFSEventSink中会实例化BucketWriter和HDFSWriter：
if (bucketWriter == null) {
HDFSWriter hdfsWriter = writerFactory.getWriter(fileType ); //获取HDFSWriter 对象
....
bucketWriter = new BucketWriter(rollInterval , rollSize , rollCount ,
batchSize, context , realPath, realName, inUsePrefix, inUseSuffix,
suffix, codeC, compType, hdfsWriter, timedRollerPool,
proxyTicket, sinkCounter , idleTimeout , idleCallback, lookupPath); //根据HDFSWriter 对象获取BucketWriter对象这里获取HDFSWriter 对象时用到了org.apache.flume.sink.hdfs.HDFSWriterFactory的getWriter方法，根据hdfs.fileType的设置会返回具体的org.apache.flume.sink.hdfs.HDFSWriter实现类的对象
目前只支持3种
static final String SequenceFileType = "SequenceFile" ;
static final String DataStreamType = "DataStream" ;
static final String CompStreamType = "CompressedStream" ;
....
public HDFSWriter getWriter(String fileType) throws IOException {
if (fileType.equalsIgnoreCase( SequenceFileType)) { //SequenceFile,sequencefile
return new HDFSSequenceFile();
} else if (fileType.equalsIgnoreCase(DataStreamType)) { //DataStream
return new HDFSDataStream();
} else if (fileType.equalsIgnoreCase(CompStreamType)) { //CompressedStream
return new HDFSCompressedDataStream();
} else {
throw new IOException("File type " + fileType + " not supported");
}BucketWriter可以理解成是对下层数据操作的一个封装，比如数据写入时其实调用了其append方法，append主要有下面几个步骤：
1）首先判断文件是否打开：
if (! isOpen) {
if(idleClosed) {
throw new IOException("This bucket writer was closed due to idling and this handle " +
"is thus no longer valid");
}
open(); //如果没有打开，则调用open->doOpen->HDFSWriter.open方法打开bucketPath (bucketPath是临时写入目录，比如tmp结尾的目录，targetPath是最终目录)
}doOpen的主要步骤
a.设置两个文件名：
bucketPath = filePath + DIRECTORY_DELIMITER + inUsePrefix + fullFileName + inUseSuffix; targetPath = filePath + DIRECTORY_DELIMITER + fullFileName;
b.调用HDFSWriter.open方法打开bucketPath
if (codeC == null) {
// Need to get reference to FS using above config before underlying
// writer does in order to avoid shutdown hook & IllegalStateExceptions
fileSystem = new Path(bucketPath ).getFileSystem(config);
LOG.info("Creating " + bucketPath );
writer.open( bucketPath);
} else {
// need to get reference to FS before writer does to avoid shutdown hook
fileSystem = new Path(bucketPath ).getFileSystem(config);
LOG.info("Creating " + bucketPath );
writer.open( bucketPath, codeC , compType );
}c.如果设置了rollInterval ,则执行计划任务调用close方法
// if time-based rolling is enabled, schedule the roll
if (rollInterval > 0) {
Callable<Void> action = new Callable<Void>() {
public Void call() throws Exception {
LOG.debug("Rolling file ({}): Roll scheduled after {} sec elapsed." ,
bucketPath, rollInterval );
try {
close();
} catch(Throwable t) {
LOG.error("Unexpected error" , t);
}
return null ;
}
};
timedRollFuture = timedRollerPool.schedule(action, rollInterval ,
TimeUnit. SECONDS);
}2）判断文件是否需要翻转（达到hdfs.rollSize或者hdfs.rollCount设置）：
// check if it‘s time to rotate the file
if (shouldRotate()) {
close(); //close调用flush+doClose，flush调用doFlush,doFlush调用HDFSWriter.sync方法把数据同步到hdfs中
open();
}其中shouldRotate(基于数量和大小的roll方式)：
private boolean shouldRotate() {
boolean doRotate = false;
if (( rollCount > 0) && (rollCount <= eventCounter )) { //hdfs.rollCount大于0并且处理的event的数量大于或等于hdfs.rollCount，doRotate 设置为true
LOG.debug( "rolling: rollCount: {}, events: {}" , rollCount , eventCounter );
doRotate = true;
}
if (( rollSize > 0) && ( rollSize <= processSize)) { //hdfs.rollCount大于0并且处理的event的数量大于或等于hdfs.rollCount，doRotate 设置为true
LOG.debug( "rolling: rollSize: {}, bytes: {}" , rollSize , processSize );
doRotate = true;
}
return doRotate;
}其中doClose主要的步骤
a.调用HDFSWriter.close方法
b.调用renameBucket方法把tmp文件命名为最终文件：
if (bucketPath != null && fileSystem != null) {
renameBucket(); // could block or throw IOException
fileSystem = null;
}其中renameBucket：
fileSystem.rename(srcPath, dstPath)
3）调用HDFSWriter.append方法写入Event
writer.append(event);
4) 更新计数器
// update statistics processSize += event.getBody(). length; eventCounter++; batchCounter++;
5）判断是否需要flush（达到hdfs.batchSize的设置），batch写入数据到hdfs
if (batchCounter == batchSize) {
flush();
}Event写入时BucketWriter的append方法调用org.apache.flume.sink.hdfs.HDFSWriter实现类的append方法，比如这里的HDFSDataStream类，HDFSDataStream的主要方法：
configure用于设置serializer：
public void configure(Context context) {
serializerType = context.getString( "serializer", "TEXT" ); //默认序列化方式为TEXT
useRawLocalFileSystem = context.getBoolean( "hdfs.useRawLocalFileSystem",
false);
serializerContext =
new Context(context.getSubProperties(EventSerializer.CTX_PREFIX));
logger.info( "Serializer = " + serializerType + ", UseRawLocalFileSystem = "
+ useRawLocalFileSystem);
}
append方法用于Event的写入，调用EventSerializer.write方法：
public void append(Event e) throws IOException {
// shun flumeformatter...
serializer.write(e); //调用EventSerializer.write方法写入Event
}open方法主要步骤：
1）根据hdfs.append.support的设置（默认为false）打开或者新建文件
boolean appending = false;
if (conf.getBoolean( "hdfs.append.support", false ) == true && hdfs.isFile
(dstPath)) { //默认hdfs.append.support为false
outStream = hdfs.append(dstPath);
appending = true;
} else {
outStream = hdfs.create(dstPath); //如果不支持append，则创建文件
}2）使用EventSerializerFactory.getInstance方法创建EventSerializer的对象
serializer = EventSerializerFactory.getInstance( serializerType, serializerContext , outStream ); //实例化EventSerializer对象
3)如果EventSerializer对象支持reopen,并且hdfs.append.support设置为true时会抛出异常
if (appending && ! serializer.supportsReopen()) {
outStream.close();
serializer = null;
throw new IOException("serializer (" + serializerType +
") does not support append");
}4）调用文件打开或者reopen之后的操作
if (appending) {
serializer.afterReopen();
} else {
serializer.afterCreate();
}
}这里hdfs.writeFormat的3种设置和对应的类：
TEXT(BodyTextEventSerializer.Builder. class), //支持reopen HEADER_AND_TEXT(HeaderAndBodyTextEventSerializer.Builder. class), //支持reopen AVRO_EVENT(FlumeEventAvroEventSerializer.Builder. class), // 不支持reopen
默认设置为TEXT，即BodyTextEventSerializer类：
private BodyTextEventSerializer(OutputStream out, Context ctx) { //构造方法
this. appendNewline = ctx.getBoolean(APPEND_NEWLINE , APPEND_NEWLINE_DFLT ); //默认为true
this. out = out;
}
....
public void write(Event e) throws IOException { //write方法
out.write(e.getBody()); //java.io.OutputStream.write，只写Event的body
if (appendNewline) { //每一行之后增加一个回车
out.write(‘\n‘);
}本文出自 “菜光光的博客” 博客，请务必保留此出处http://caiguangguang.blog.51cto.com/1652935/1618343
原文地址：http://caiguangguang.blog.51cto.com/1652935/1618343
