Flume-NG源码阅读之HDFSEventSink

　1 public void configure(Context context) {  
　2     this.context = context;
  3     //HDFS目录路径，必需(eg hdfs://namenode/flume/webdata/)
  4     filePath = Preconditions.checkNotNull(
  5         context.getString("hdfs.path"), "hdfs.path is required");
  6     //在Hdfs目录中生成的文件名字的前缀
  7     fileName = context.getString("hdfs.filePrefix", defaultFileName);
  8     //文件后缀，例如.avro，一般不用
  9     this.suffix = context.getString("hdfs.fileSuffix", defaultSuffix);
 10     //内部写文件的时候表示正在写的文件的前缀和后缀
 11     inUsePrefix = context.getString("hdfs.inUsePrefix", defaultInUsePrefix);
 12     inUseSuffix = context.getString("hdfs.inUseSuffix", defaultInUseSuffix);//默认是.tmp
 13     String tzName = context.getString("hdfs.timeZone");
 14     timeZone = tzName == null ? null : TimeZone.getTimeZone(tzName);
 15     //当前写入的文件滚动间隔，默认30秒生成一个新的文件，0表示不基于时间间隔来滚动
 16     rollInterval = context.getLong("hdfs.rollInterval", defaultRollInterval);
 17     //以文件大小触发文件滚动，单位字节，0表示不基于文件大小间隔来滚动
 18     rollSize = context.getLong("hdfs.rollSize", defaultRollSize);
 19     //以写入的事件数触发文件滚动， 0表示不基于事件数大小间隔来滚动
 20     rollCount = context.getLong("hdfs.rollCount", defaultRollCount);
 21     //事件刷新到HDFS之前的数量
 22     batchSize = context.getLong("hdfs.batchSize", defaultBatchSize);
 23     //超时后关闭无效文件（0 =禁止自动关闭闲置的文件）
 24     idleTimeout = context.getInteger("hdfs.idleTimeout", 0);
 25     //压缩编码类型. one of following : gzip, bzip2, lzo, snappy
 26     String codecName = context.getString("hdfs.codeC");
 27     //文件格式：当前为SequenceFile, DataStream or CompressedStream。
 28     //（1）DataStream不压缩输出文件，不能设置codeC选项，（2）CompressedStream需要设置hdfs.codeC的一个可用的编解码器
 29     fileType = context.getString("hdfs.fileType", defaultFileType);
 30     //允许打开的文件数。如果超过这个数字，最早的文件被关闭。
 31     maxOpenFiles = context.getInteger("hdfs.maxOpenFiles", defaultMaxOpenFiles);
 32     //HDFS的操作允许的毫秒数，如打开，写，刷新，关闭。这个数字应该增加，如果正在发生许多HDFS超时操作。
 33     callTimeout = context.getLong("hdfs.callTimeout", defaultCallTimeout);
 34     //每个HDFS sink用于HDFS io操作的线程数，如打开、写入等操作。
 35     threadsPoolSize = context.getInteger("hdfs.threadsPoolSize",
 36         defaultThreadPoolSize);
 37     //每个HDFS sink用于调度定时文件滚动的线程数
 38     rollTimerPoolSize = context.getInteger("hdfs.rollTimerPoolSize",
 39         defaultRollTimerPoolSize);
 40     //安全认证时使用Kerberos user principal for accessing secure HDFS
 41     kerbConfPrincipal = context.getString("hdfs.kerberosPrincipal", "");
 42     //安全认证时使用Kerberos keytab for accessing secure HDFS
 43     kerbKeytab = context.getString("hdfs.kerberosKeytab", "");
 44     proxyUserName = context.getString("hdfs.proxyUser", "");　　//代理用户
 45 
 46     Preconditions.checkArgument(batchSize > 0,
 47         "batchSize must be greater than 0");
 48     if (codecName == null) {　　//不压缩数据
 49       codeC = null;
 50       compType = CompressionType.NONE;
 51     } else {　　　　//压缩数据
 52       codeC = getCodec(codecName);
 53       // TODO : set proper compression type
 54       compType = CompressionType.BLOCK;
 55     }
 56 
 57     // Do not allow user to set fileType DataStream with codeC together
 58     // To prevent output file with compress extension (like .snappy)
 59     if(fileType.equalsIgnoreCase(HDFSWriterFactory.DataStreamType)//如果fileType是DataStream，则不允许压缩
 60         && codecName != null) {
 61       throw new IllegalArgumentException("fileType: " + fileType +
 62           " which does NOT support compressed output. Please don‘t set codeC" +
 63           " or change the fileType if compressed output is desired.");
 64     }
 65 
 66     if(fileType.equalsIgnoreCase(HDFSWriterFactory.CompStreamType)) {//如果fileType是压缩类型，则codeC不允许为空
 67       Preconditions.checkNotNull(codeC, "It‘s essential to set compress codec"
 68           + " when fileType is: " + fileType);
 69     }
 70 
 71     if (!authenticate()) {　　//认证
 72       LOG.error("Failed to authenticate!");
 73     }
 74     //时间戳是否四舍五入（如果为true，会影响所有基于时间的转义序列％t除外）
 75     needRounding = context.getBoolean("hdfs.round", false);
 76 
 77     if(needRounding) {
 78         //The unit of the round down value - second, minute or hour.
 79       String unit = context.getString("hdfs.roundUnit", "second");　　//滚动时间单位
 80       if (unit.equalsIgnoreCase("hour")) {
 81         this.roundUnit = Calendar.HOUR_OF_DAY;
 82       } else if (unit.equalsIgnoreCase("minute")) {
 83         this.roundUnit = Calendar.MINUTE;
 84       } else if (unit.equalsIgnoreCase("second")){
 85         this.roundUnit = Calendar.SECOND;
 86       } else {
 87         LOG.warn("Rounding unit is not valid, please set one of" +
 88             "minute, hour, or second. Rounding will be disabled");
 89         needRounding = false;
 90       }
 91       //Rounded down to the highest multiple of this (in the unit configured using hdfs.roundUnit), less than current time.
 92       this.roundValue = context.getInteger("hdfs.roundValue", 1);　　//滚动时间大小
 93       if(roundUnit == Calendar.SECOND || roundUnit == Calendar.MINUTE){//检查是否符合分、秒数值,0<v<=60
 94         Preconditions.checkArgument(roundValue > 0 && roundValue <= 60,
 95             "Round value" +
 96             "must be > 0 and <= 60");
 97       } else if (roundUnit == Calendar.HOUR_OF_DAY){
 98         Preconditions.checkArgument(roundValue > 0 && roundValue <= 24,　　//检查是否符合时数值0<v<=24
 99             "Round value" +
100             "must be > 0 and <= 24");
101       }
102     }
103 
104     if (sinkCounter == null) {//构造计数器
105       sinkCounter = new SinkCounter(getName());
106     }
107   }

 1 public void start() {
 2     String timeoutName = "hdfs-" + getName() + "-call-runner-%d";
 3     callTimeoutPool = Executors.newFixedThreadPool(threadsPoolSize,
 4             new ThreadFactoryBuilder().setNameFormat(timeoutName).build());　　//这个线程池用来将event写入HDFS文件
 5 
 6     String rollerName = "hdfs-" + getName() + "-roll-timer-%d";
 7     timedRollerPool = Executors.newScheduledThreadPool(rollTimerPoolSize,
 8             new ThreadFactoryBuilder().setNameFormat(rollerName).build());　　//这个线程池用来滚动文件
 9 
10     this.sfWriters = new WriterLinkedHashMap(maxOpenFiles);　　//用来存储文件的绝对路径以及对应的BucketWriter
11     sinkCounter.start();
12     super.start();
13   }

 1 public Status process() throws EventDeliveryException {
 2     Channel channel = getChannel();    //获取对应的channel
 3     Transaction transaction = channel.getTransaction();//获得事务
 4     List<BucketWriter> writers = Lists.newArrayList(); //BucketWriter列表
 5     transaction.begin();
 6     try {
 7       int txnEventCount = 0;
 8       for (txnEventCount = 0; txnEventCount < batchSize; txnEventCount++) {//批量处理
 9         Event event = channel.take();    //获取event
10         if (event == null) {
11           break;
12         }
13 
14         // reconstruct the path name by substituting place holders
15         String realPath = BucketPath.escapeString(filePath, event.getHeaders(),
16             timeZone, needRounding, roundUnit, roundValue);    //格式化后的HDFS目录
17         String realName = BucketPath.escapeString(fileName, event.getHeaders(),
18           timeZone, needRounding, roundUnit, roundValue);    //格式化后的文件名
19 
20         String lookupPath = realPath + DIRECTORY_DELIMITER + realName;    //要写入的文件的HDFS绝对路径
21         BucketWriter bucketWriter = sfWriters.get(lookupPath);    //获取文件的BucketWriter
22 
23         // we haven‘t seen this file yet, so open it and cache the handle
24         if (bucketWriter == null) {    //如果没有这个文件
25             //根据fileType类型构造HDFSWriter(三种：SequenceFile、DataStream、CompressedStream)
26           HDFSWriter hdfsWriter = writerFactory.getWriter(fileType);    
27 
28           WriterCallback idleCallback = null;
29           if(idleTimeout != 0) {
30             idleCallback = new WriterCallback() {
31               @Override
32               public void run(String bucketPath) {
33                 sfWriters.remove(bucketPath);
34               }
35             };
36           }
37           bucketWriter = new BucketWriter(rollInterval, rollSize, rollCount,
38               batchSize, context, realPath, realName, inUsePrefix, inUseSuffix,
39               suffix, codeC, compType, hdfsWriter, timedRollerPool,
40               proxyTicket, sinkCounter, idleTimeout, idleCallback, lookupPath);
41 
42           sfWriters.put(lookupPath, bucketWriter);    //将文件路径和BucketWriter组成K-V，放入sfWriters
43         }
44 
45         // track the buckets getting written in this transaction
46         if (!writers.contains(bucketWriter)) {//如果BucketWriter列表没有正在写的文件——bucketWriter，则加入
47           writers.add(bucketWriter);
48         }
49 
50         // Write the data to HDFS
51         append(bucketWriter, event);    //将event写入bucketWriter对应的文件中
52       }
53 
54       if (txnEventCount == 0) {    //这次事务没有处理任何event
55         sinkCounter.incrementBatchEmptyCount();
56       } else if (txnEventCount == batchSize) {//一次处理batchSize个event
57         sinkCounter.incrementBatchCompleteCount();
58       } else {//channel中剩余的events不足batchSize
59         sinkCounter.incrementBatchUnderflowCount();
60       }
61 
62       // flush all pending buckets before committing the transaction
63       for (BucketWriter bucketWriter : writers) {    //将所有BucketWriter数据刷新到HDFS中
64         flush(bucketWriter);
65       }
66 
67       transaction.commit();    //提交事务
68 
69       if (txnEventCount < 1) {
70         return Status.BACKOFF;
71       } else {
72         sinkCounter.addToEventDrainSuccessCount(txnEventCount);
73         return Status.READY;
74       }
75     } catch (IOException eIO) {
76       transaction.rollback();//异常后回滚
77       LOG.warn("HDFS IO error", eIO);
78       return Status.BACKOFF;
79     } catch (Throwable th) {
80       transaction.rollback();//异常后回滚
81       LOG.error("process failed", th);
82       if (th instanceof Error) {
83         throw (Error) th;
84       } else {
85         throw new EventDeliveryException(th);
86       }
87     } finally {
88       transaction.close();//关闭事务
89     }
90   }

 1 public synchronized void append(Event event)
 2           throws IOException, InterruptedException {
 3     checkAndThrowInterruptedException();
 4     if (!isOpen) {
 5       if(idleClosed) {
 6         throw new IOException("This bucket writer was closed due to idling and this handle " +
 7             "is thus no longer valid");
 8       }
 9       open();//已经写完一个文件，新建新文件
10     }
11 
12     // check if it‘s time to rotate the file
13     if (shouldRotate()) {//检查行数、大小是否改完成一个文件
14       close();
15       open();//新建新文件
16     }
17 
18     // write the event
19     try {
20       sinkCounter.incrementEventDrainAttemptCount();
21       writer.append(event); // could block写数据
22     } catch (IOException e) {
23       LOG.warn("Caught IOException writing to HDFSWriter ({}). Closing file (" +
24           bucketPath + ") and rethrowing exception.",
25           e.getMessage());
26       try {
27         close();
28       } catch (IOException e2) {
29         LOG.warn("Caught IOException while closing file (" +
30              bucketPath + "). Exception follows.", e2);
31       }
32       throw e;
33     }
34 
35     // update statistics
36     processSize += event.getBody().length;
37     eventCounter++;
38     batchCounter++;
39 
40     if (batchCounter == batchSize) {
41       flush();
42     }
43   }