code: @SuppressWarnings("rawtypes")
      @Override
      public void prepare(Map stormConf, TopologyContext context, OutputCollector collector)
      {
          super.prepare(stormConf, context, collector);
  
          String componentId = context.getThisComponentId();
          int taskId = context.getThisTaskId();
  
          HdfsStoreConfig config = new HdfsStoreConfig();
  
          config.setOutputUri((String) stormConf.get("hdfsstorebolt.outputuri"));
          config.setFileNameHeader((String) stormConf.get("hdfsstorebolt.filenameheader"));
          config.setFileSwitchIntarval(((Long) stormConf.get("hdfsstorebolt.interval")).intValue());
          config.setFileNameBody("_" + componentId + "_" + taskId + "_");
  
          boolean isPreprocess = true;
          Object isPreprocessObj = stormConf.get("hdfsstorebolt.executepreprocess");
          if (isPreprocessObj != null && isPreprocessObj instanceof Boolean)
          {
              isPreprocess = ((Boolean) isPreprocessObj).booleanValue();
          }
  
          try
          {
              // HDFSファイルシステム取得
              Configuration conf = new Configuration();
              Path dstPath = new Path(config.getOutputUri());
              FileSystem fileSystem = dstPath.getFileSystem(conf);
  
              // HDFSに対する前処理実施。一時ファイルを本ファイルにリネームする。
              if (isPreprocess)
              {
                  HdfsPreProcessor.execute(fileSystem, config.getOutputUri(),
                          config.getFileNameHeader() + config.getFileNameBody(),
                          config.getTmpFileSuffix());
              }
  
              this.delegate = new HdfsOutputSwitcher();
              this.delegate.initialize(fileSystem, config, System.currentTimeMillis());
          }
          catch (Exception ex)
          {
              logger.warn("Failed to HDFS write initialize.", ex);
              throw new InitFailException(ex);
          }
      }
docs: {@inheritDoc}