class HiveOutputFormatFactory implements OutputFormatFactory<Row> {

	@Override
	public HiveOutputFormat createOutputFormat(Path outPath) {
		try {
			if (!inited) {
				init();
				inited = true;
			}

			JobConf conf = new JobConf(confWrapper.conf());

			final boolean isCompressed = conf.getBoolean(HiveConf.ConfVars.COMPRESSRESULT.varname, false);
			if (isCompressed) {
				String codecStr = conf.get(HiveConf.ConfVars.COMPRESSINTERMEDIATECODEC.varname);
				if (!StringUtils.isNullOrWhitespaceOnly(codecStr)) {
					//noinspection unchecked
					Class<? extends CompressionCodec> codec =
							(Class<? extends CompressionCodec>) Class.forName(codecStr, true,
									Thread.currentThread().getContextClassLoader());
					FileOutputFormat.setOutputCompressorClass(conf, codec);
				}
				String typeStr = conf.get(HiveConf.ConfVars.COMPRESSINTERMEDIATETYPE.varname);
				if (!StringUtils.isNullOrWhitespaceOnly(typeStr)) {
					SequenceFile.CompressionType style = SequenceFile.CompressionType.valueOf(typeStr);
					SequenceFileOutputFormat.setOutputCompressionType(conf, style);
				}
			}

			RecordWriter recordWriter = hiveShim.getHiveRecordWriter(
					conf,
					outputFormat,
					recordSerDe.getSerializedClass(),
					isCompressed,
					tableProperties,
					HadoopFileSystem.toHadoopPath(outPath));
			return new HiveOutputFormat(recordWriter);
		} catch (Exception e) {
			throw new FlinkHiveException(e);
		}
	}

	public  HiveOutputFormatFactory(
			JobConf jobConf,
			String outputFormat,
			SerDeInfo serDeInfo,
			TableSchema schema,
			String[] partitionColumns,
			Properties tableProperties,
			HiveShim hiveShim);

}