code: @SuppressWarnings({"rawtypes"})
      @Override
      public void open(Map conf, TopologyContext context)
      {
          this.taskIndex = context.getThisTaskIndex();
          this.fileName = this.baseFileName + "_" + this.taskIndex;
          File targetFile = new File(this.dataFilePath, this.fileName);
          try
          {
              this.fileContents = FileUtils.readLines(targetFile);
              this.fileContentsSize = this.fileContents.size();
          }
          catch (IOException ex)
          {
              // 読込に失敗した場合は例外を投げてフェールオーバーさせる。
              throw new RuntimeException(ex);
          }
      }
docs: {@inheritDoc}