@Override
  public RecordWriter<WritableComparable<?>, HCatRecord> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
    //this needs to be manually set, under normal circumstances MR Task does this
    setWorkOutputPath(context);

    //Configure the output key and value classes.
    // This is required for writing null as key for file based tables.
    context.getConfiguration().set("mapred.output.key.class",
      NullWritable.class.getName());
    String jobInfoString = context.getConfiguration().get(
      HCatConstants.HCAT_KEY_OUTPUT_INFO);
    OutputJobInfo jobInfo = (OutputJobInfo) HCatUtil
      .deserialize(jobInfoString);
    StorerInfo storeInfo = jobInfo.getTableInfo().getStorerInfo();
    HiveStorageHandler storageHandler = HCatUtil.getStorageHandler(
      context.getConfiguration(), storeInfo);
    Class<? extends AbstractSerDe> serde = storageHandler.getSerDeClass();
    AbstractSerDe sd = (AbstractSerDe) ReflectionUtils.newInstance(serde,
      context.getConfiguration());
    context.getConfiguration().set("mapred.output.value.class",
      sd.getSerializedClass().getName());

    RecordWriter<WritableComparable<?>, HCatRecord> rw;
    if (HCatBaseOutputFormat.getJobInfo(context.getConfiguration()).isDynamicPartitioningUsed()){
      // When Dynamic partitioning is used, the RecordWriter instance initialized here isn't used. Can use null.
      // (That's because records can't be written until the values of the dynamic partitions are deduced.
      // By that time, a new local instance of RecordWriter, with the correct output-path, will be constructed.)
      rw = new DynamicPartitionFileRecordWriterContainer(
          (org.apache.hadoop.mapred.RecordWriter)null, context);
    } else {
      Path parentDir = new Path(context.getConfiguration().get("mapred.work.output.dir"));
      String extension = HiveConf.getVar(context.getConfiguration(), HiveConf.ConfVars.OUTPUT_FILE_EXTENSION,"");
      Path childPath = new Path(parentDir,FileOutputFormat.getUniqueName(new JobConf(context.getConfiguration()),
               context.getConfiguration().get("mapreduce.output.basename", "part")) + extension);

      rw = new StaticPartitionFileRecordWriterContainer(
          getBaseOutputFormat().getRecordWriter(
              parentDir.getFileSystem(context.getConfiguration()),
              new JobConf(context.getConfiguration()),
              childPath.toString(),
              InternalUtil.createReporter(context)),
          context);
    }
    return rw;
  }
@param of base OutputFormat to contain
hcatalog/core/src/main/java/org/apache/hive/hcatalog/mapreduce/FileOutputFormatContainer.java