前言：HDFS sink原生的解析時間戳的程式碼效能不高，可以通過修改原始碼提升效能。具體操作參考連結：http://www.cnblogs.com/lxf20061900/p/4014281.html

HDFS sink常用配置項：

一般使用hdfs sink都會採用滾動生成檔案的方式，hdfs sink滾動生成檔案的策略有：

• 基於時間
• 基於檔案大小
• 基於hdfs檔案副本數（一般要規避這種情況）
• 基於event數量
• 基於檔案閒置時間

1、基於時間策略

配置項：hdfs.rollInterval

預設值：30秒

說明：如果設定為0表示禁用這個策略

原理： 在org.apache.flume.sink.hdfs.BucketWriter.append方法中開啟一個檔案，都會呼叫open方法，如果設定了hdfs.rollInterval，那麼hdfs.rollInterval秒之內只要其他策略沒有關閉檔案，檔案會在hdfs.rollInterval秒之後關閉。

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 {
        // Roll the file and remove reference from sfWriters map.
        close(true);
      } catch (Throwable t) {
        LOG.error("Unexpected error", t);
      }
      return null;
    }
  };
  timedRollFuture = timedRollerPool.schedule(action, rollInterval,
      TimeUnit.SECONDS);

}

2、基於檔案大小和event數量策略

配置項：

檔案大小策略：hdfs.rollSize

event數量策略：hdfs.rollCount

預設值：

檔案大小策略：1024位元組

event數量策略：10

說明：如果設定為0表示禁用這些策略

原理： 這2種策略都是在org.apache.flume.sink.hdfs.BucketWriter.shouldRotate方法中進行判斷的，只要doRotate的值為true，那麼當前檔案就會關閉，即滾動到下一個檔案。

private boolean shouldRotate() {

boolean doRotate = false;

if (writer.isUnderReplicated()) {
  this.isUnderReplicated = true;
  doRotate = true;
} else {
  this.isUnderReplicated = false;
}

if ((rollCount > 0) && (rollCount <= eventCounter)) {
  LOG.debug("rolling: rollCount: {}, events: {}", rollCount, eventCounter);
  doRotate = true;
}

if ((rollSize > 0) && (rollSize <= processSize)) {
  LOG.debug("rolling: rollSize: {}, bytes: {}", rollSize, processSize);
  doRotate = true;
}

return doRotate;

}

注意：如果同時配置了時間策略和檔案大小策略，那麼會先判斷時間，如果時間沒到再判斷其他的條件。

3、基於hdfs檔案副本數

配置項：hdfs.minBlockReplicas

預設值：和hdfs的副本數一致

原理： 從上面的程式碼中可以看到，判斷副本數的關鍵方法是writer.isUnderReplicated()，即

public boolean isUnderReplicated() {

try {
  int numBlocks = getNumCurrentReplicas();
  if (numBlocks == -1) {
    return false;
  }
  int desiredBlocks;
  if (configuredMinReplicas != null) {
    desiredBlocks = configuredMinReplicas;
  } else {
    desiredBlocks = getFsDesiredReplication();
  }
  return numBlocks < desiredBlocks;
} catch (IllegalAccessException e) {
  logger.error("Unexpected error while checking replication factor", e);
} catch (InvocationTargetException e) {
  logger.error("Unexpected error while checking replication factor", e);
} catch (IllegalArgumentException e) {
  logger.error("Unexpected error while checking replication factor", e);
}
return false;

}

也就是說，如果當前正在寫的檔案的副本數小於hdfs.minBlockReplicas，此方法返回true，其他情況都返回false。假設這個方法返回true，那麼看一下會發生什麼事情。

首先就是上面程式碼提到的shouldRotate方法肯定返回的是true。再繼續跟蹤，下面的程式碼是關鍵

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();
  }
}

這裡maxConsecUnderReplRotations是固定的值30，也就是說，檔案滾動生成了30個之後，就不會再滾動了，因為將doRotate設定為了false。所以，從這裡可以看到，如果isUnderReplicated方法返回的是true，可能會導致檔案的滾動和預期的不一致。規避這個問題的方法就是將hdfs.minBlockReplicas設定為1，一般hdfs的副本數肯定都是大於等於1的，所以isUnderReplicated方法一定會返回false。 所以一般情況下，要規避這種情況，避免影響檔案的正常滾動。

4、基於檔案閒置時間策略

配置項：hdfs.idleTimeout

預設值：0

說明：預設啟動這個功能

這種策略很簡單，如果檔案在hdfs.idleTimeout秒的時間裡都是閒置的，沒有任何資料寫入，那麼當前檔案關閉，滾動到下一個檔案。

public synchronized void flush() throws IOException, InterruptedException {

checkAndThrowInterruptedException();
if (!isBatchComplete()) {
  doFlush();

  if (idleTimeout > 0) {
    // if the future exists and couldn't be cancelled, that would mean it has already run
    // or been cancelled
    if (idleFuture == null || idleFuture.cancel(false)) {
      Callable<Void> idleAction = new Callable<Void>() {
        public Void call() throws Exception {
          LOG.info("Closing idle bucketWriter {} at {}", bucketPath,
                   System.currentTimeMillis());
          if (isOpen) {
            close(true);
          }
          return null;
        }
      };
      idleFuture = timedRollerPool.schedule(idleAction, idleTimeout,
          TimeUnit.SECONDS);
    }
  }
}

}

以上滾動生成檔案的內容，參考連結 https://www.jianshu.com/p/4f43780c82e9