@Override
                public void run() {
                    try {
                        BufferedReader reader;
                        reader = new BufferedReader(new InputStreamReader(hboxProcess.getInputStream()));
                        List<OutputInfo> outputs = Arrays.asList(amClient.getOutputLocation());
                        JobConf jobConf = new JobConf(conf);
                        jobConf.setOutputKeyClass(Text.class);
                        jobConf.setOutputValueClass(Text.class);
                        jobConf.setBoolean("mapred.output.compress", true);
                        jobConf.set("mapred.output.compression.codec", "org.apache.hadoop.io.compress.GzipCodec");
                        jobConf.setOutputFormat(TextOutputFormat.class);
                        Path remotePath =
                                new Path(outputs.get(0).getDfsLocation() + "/_temporary/" + containerId.toString());
                        FileSystem dfs = remotePath.getFileSystem(jobConf);
                        // FileOutputFormat.setOutputPath(jobConf, remotePath.makeQualified(dfs));
                        jobConf.set(
                                HboxConstants.STREAM_OUTPUT_DIR,
                                remotePath.makeQualified(dfs).toString());

                        OutputFormat outputFormat = ReflectionUtils.newInstance(
                                conf.getClass(
                                        HboxConfiguration2.HBOX_OUTPUTFORMAT_CLASS,
                                        HboxConfiguration2.DEFAULT_HBOX_OUTPUTF0RMAT_CLASS,
                                        OutputFormat.class),
                                jobConf);
                        outputFormat.checkOutputSpecs(dfs, jobConf);
                        JobID jobID = new JobID(new SimpleDateFormat("yyyyMMddHHmm").format(new Date()), 0);
                        TaskAttemptID taId = new TaskAttemptID(new TaskID(jobID, true, 0), 0);
                        jobConf.set("mapred.tip.id", taId.getTaskID().toString());
                        jobConf.set("mapred.task.id", taId.toString());
                        jobConf.set("mapred.job.id", jobID.toString());
                        amClient.reportMapedTaskID(containerId, taId.toString());
                        RecordWriter writer = outputFormat.getRecordWriter(dfs, jobConf, "part-r", Reporter.NULL);
                        String hboxStreamResultLine;
                        while ((hboxStreamResultLine = reader.readLine()) != null) {
                            // LOG.info(hboxStreamResultLine);
                            writer.write(null, hboxStreamResultLine);
                            // reader.skip(1);
                        }
                        writer.close(Reporter.NULL);
                        reader.close();
                    } catch (Exception e) {
                        LOG.warn("Exception in thread stdoutRedirectThread");
                        e.printStackTrace();
                    }
                }
set TF_CLUSTER_DEF in env
python script can load cluster def use "json.loads(os.environ["TF_CLUSTER_DEF"])"
core/src/main/java/net/qihoo/hbox/container/HboxContainer.java