上一篇说了HDFSEventSink的实现，这里根据hdfs sink的配置和调用分析来看下sink中整个hdfs数据写入的过程：
线上hdfs sink的几个重要设置| 1 2 3 4 5 6 7 8 | 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的序列化方式 上面的设置，其数据写入流程大概如下：| 1 | SinkRunner.process->SinkProcessor.process->HDFSEventSink.process->HDFSEventSink.append->BucketWriter.append->HDFSWriter.append->HDFSDataStream.append->BodyTextEventSerializer.write->java.io.OutputStream.write |
简单说下：
在HDFSEventSink中会实例化BucketWriter和HDFSWriter：| 1 2 3 4 5 6 7 | 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种| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 | 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）首先判断文件是否打开：| 1 2 3 4 5 6 7 | 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.设置两个文件名：| 1 2 3 | bucketPath = filePath + DIRECTORY_DELIMITER + inUsePrefix + fullFileName + inUseSuffix; targetPath = filePath + DIRECTORY_DELIMITER + fullFileName; |
b.调用HDFSWriter.open方法打开bucketPath
| 1 2 3 4 5 6 7 8 9 10 11 12 | 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方法
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 | // 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设置）：
| 1 2 3 4 5 | // check if it's time to rotate the file if (shouldRotate()) { close(); //close调用flush+doClose，flush调用doFlush,doFlush调用HDFSWriter.sync方法把数据同步到hdfs中 open(); } |
其中shouldRotate(基于数量和大小的roll方式)：
| 1 2 3 4 5 6 7 8 9 10 11 12 | 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文件命名为最终文件：| 1 2 3 4 | if (bucketPath != null && fileSystem != null ) { renameBucket(); // could block or throw IOException fileSystem = null ; } |
其中renameBucket：
| 1 | fileSystem.rename(srcPath, dstPath) |
3）调用HDFSWriter.append方法写入Event
| 1 | writer.append(event); |
4) 更新计数器
| 1 2 3 4 | // update statistics processSize += event.getBody(). length; eventCounter++; batchCounter++; |
5）判断是否需要flush（达到hdfs.batchSize的设置），batch写入数据到hdfs
| 1 2 3 | if (batchCounter == batchSize) { flush(); } |
Event写入时BucketWriter的append方法调用org.apache.flume.sink.hdfs.HDFSWriter实现类的append方法，比如这里的HDFSDataStream类，HDFSDataStream的主要方法：
configure用于设置serializer：| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 | 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）打开或者新建文件| 1 2 3 4 5 6 7 8 | 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的对象
| 1 2 | serializer = EventSerializerFactory.getInstance( serializerType, serializerContext , outStream ); //实例化EventSerializer对象 |
3)如果EventSerializer对象支持reopen,并且hdfs.append.support设置为true时会抛出异常
| 1 2 3 4 5 6 | if (appending && ! serializer.supportsReopen()) { outStream.close(); serializer = null ; throw new IOException( "serializer (" + serializerType + ") does not support append" ); } |
4）调用文件打开或者reopen之后的操作
| 1 2 3 4 5 6 | if (appending) { serializer.afterReopen(); } else { serializer.afterCreate(); } } |
这里hdfs.writeFormat的3种设置和对应的类：
| 1 2 3 | TEXT(BodyTextEventSerializer.Builder. class ), //支持reopen HEADER_AND_TEXT(HeaderAndBodyTextEventSerializer.Builder. class ), //支持reopen AVRO_EVENT(FlumeEventAvroEventSerializer.Builder. class ), // 不支持reopen |
默认设置为TEXT，即BodyTextEventSerializer类：
| 1 2 3 4 5 6 7 8 9 10 | 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' ); } |