import org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy; //导入依赖的package包/类
public static FileRotationPolicy getTimeBasedFileRotationPolicy(String rotationUnits, int rotationCount) {
    TimedRotationPolicy.TimeUnit units;
    if (rotationUnits.toUpperCase().equals("SECONDS")) {
        units = TimedRotationPolicy.TimeUnit.SECONDS;
    } else if (rotationUnits.toUpperCase().equals("MINUTES")) {
        units = TimedRotationPolicy.TimeUnit.MINUTES;
    } else if (rotationUnits.toUpperCase().equals("HOURS")) {
        units = TimedRotationPolicy.TimeUnit.HOURS;
    } else if (rotationUnits.toUpperCase().equals("DAYS")) {
        units = TimedRotationPolicy.TimeUnit.DAYS;
    } else {
        units = TimedRotationPolicy.TimeUnit.MINUTES;
    }

    return new TimedRotationPolicy(rotationCount, units);
}
 

import org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy; //导入依赖的package包/类
private static FileRotationPolicy getTimeBasedFileRotationPolicy(String unitsConfigured, int timeConfigured) {
    TimedRotationPolicy.TimeUnit units;
    if (unitsConfigured.toUpperCase().equals("SECONDS")) {
        units = TimedRotationPolicy.TimeUnit.SECONDS;
    } else if (unitsConfigured.toUpperCase().equals("MINUTES")) {
        units = TimedRotationPolicy.TimeUnit.MINUTES;
    } else if (unitsConfigured.toUpperCase().equals("HOURS")) {
        units = TimedRotationPolicy.TimeUnit.HOURS;
    } else if (unitsConfigured.toUpperCase().equals("DAYS")) {
        units = TimedRotationPolicy.TimeUnit.DAYS;
    } else {
        units = TimedRotationPolicy.TimeUnit.MINUTES;
    }

    return new TimedRotationPolicy(timeConfigured, units);
}
 

import org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy; //导入依赖的package包/类
private void initialize() throws IOException {
  this.fs = FileSystem.get(new Configuration());
  this.currentFile = createOutputFile();
  if(this.rotationPolicy instanceof TimedRotationPolicy){
    long interval = ((TimedRotationPolicy)this.rotationPolicy).getInterval();
    this.rotationTimer = new Timer(true);
    TimerTask task = new TimerTask() {
      @Override
      public void run() {
        try {
          rotateOutputFile();
        } catch(IOException e){
          LOG.warn("IOException during scheduled file rotation.", e);
        }
      }
    };
    this.rotationTimer.scheduleAtFixedRate(task, interval, interval);
  }
}
 

import org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
    Config config = new Config();
    config.setNumWorkers(1);

    SentenceSpout spout = new SentenceSpout();

    // sync the filesystem after every 1k tuples
    SyncPolicy syncPolicy = new CountSyncPolicy(1000);

    // rotate files when they reach 5MB
    FileRotationPolicy rotationPolicy = new TimedRotationPolicy(1.0f, TimedRotationPolicy.TimeUnit.MINUTES);

    FileNameFormat fileNameFormat = new DefaultFileNameFormat()
            .withPath("/foo/")
            .withExtension(".txt");



    // use "|" instead of "," for field delimiter
    RecordFormat format = new DelimitedRecordFormat()
            .withFieldDelimiter("|");

    Yaml yaml = new Yaml();
    InputStream in = new FileInputStream(args[1]);
    Map<String, Object> yamlConf = (Map<String, Object>) yaml.load(in);
    in.close();
    config.put("hdfs.config", yamlConf);

    HdfsBolt bolt = new HdfsBolt()
            .withConfigKey("hdfs.config")
            .withFsUrl(args[0])
            .withFileNameFormat(fileNameFormat)
            .withRecordFormat(format)
            .withRotationPolicy(rotationPolicy)
            .withSyncPolicy(syncPolicy)
            .addRotationAction(new MoveFileAction().toDestination("/dest2/"));

    TopologyBuilder builder = new TopologyBuilder();

    builder.setSpout(SENTENCE_SPOUT_ID, spout, 1);
    // SentenceSpout --> MyBolt
    builder.setBolt(BOLT_ID, bolt, 4)
            .shuffleGrouping(SENTENCE_SPOUT_ID);

    if (args.length == 2) {
        LocalCluster cluster = new LocalCluster();

        cluster.submitTopology(TOPOLOGY_NAME, config, builder.createTopology());
        waitForSeconds(120);
        cluster.killTopology(TOPOLOGY_NAME);
        cluster.shutdown();
        System.exit(0);
    } else if (args.length == 3) {
        StormSubmitter.submitTopology(args[0], config, builder.createTopology());
    } else{
        System.out.println("Usage: HdfsFileTopology [topology name] <yaml config file>");
    }
}