@Override
                public void run() {
                    try {
                        OutputStreamWriter osw = new OutputStreamWriter(hboxProcess.getOutputStream());
                        File gzFile = new File("inputformatCache.gz");
                        GZIPOutputStream gos = new GZIPOutputStream(new FileOutputStream(gzFile));
                        boolean isCache = conf.getBoolean(
                                HboxConfiguration.HBOX_INPUTFORMAT_CACHE,
                                HboxConfiguration.DEFAULT_HBOX_INPUTFORMAT_CACHE);
                        List<InputSplit> inputs = Arrays.asList(amClient.getStreamInputSplit(containerId));
                        JobConf jobConf = new JobConf(conf);
                        RecordReader reader;
                        InputFormat inputFormat = ReflectionUtils.newInstance(
                                conf.getClass(
                                        HboxConfiguration2.HBOX_INPUTF0RMAT_CLASS,
                                        HboxConfiguration2.DEFAULT_HBOX_INPUTF0RMAT_CLASS,
                                        InputFormat.class),
                                jobConf);
                        for (int j = 0;
                                j
                                        < conf.getInt(
                                                HboxConfiguration.HBOX_STREAM_EPOCH,
                                                HboxConfiguration.DEFAULT_HBOX_STREAM_EPOCH);
                                j++) {
                            LOG.info("Epoch " + (j + 1) + " starting...");
                            for (int i = 0, len = inputs.size(); i < len; i++) {
                                LOG.info("split " + (i + 1) + " is handling...");
                                reader = inputFormat.getRecordReader(inputs.get(i), jobConf, Reporter.NULL);
                                Object key = reader.createKey();
                                Object value = reader.createValue();
                                Boolean finished = false;
                                while (!finished) {
                                    try {
                                        finished = !reader.next(key, value);
                                        if (finished) {
                                            break;
                                        }
                                        osw.write(value.toString());
                                        osw.write("\n");
                                        if (j == 0 && isCache) {
                                            if (conf.getInt(
                                                            HboxConfiguration.HBOX_STREAM_EPOCH,
                                                            HboxConfiguration.DEFAULT_HBOX_STREAM_EPOCH)
                                                    > 1) {
                                                gos.write(value.toString().getBytes());
                                                gos.write("\n".getBytes());
                                                if ((gzFile.length() / 1024 / 1024)
                                                        > conf.getInt(
                                                                HboxConfiguration.HBOX_INPUTFORMAT_CACHESIZE_LIMIT,
                                                                HboxConfiguration
                                                                        .DEFAULT_HBOX_INPUTFORMAT_CACHESIZE_LIMIT)) {
                                                    LOG.info("Inputformat cache file size is:"
                                                            + gzFile.length() / 1024 / 1024 + "M "
                                                            + "beyond the limit size:"
                                                            + conf.getInt(
                                                                    HboxConfiguration.HBOX_INPUTFORMAT_CACHESIZE_LIMIT,
                                                                    HboxConfiguration
                                                                            .DEFAULT_HBOX_INPUTFORMAT_CACHESIZE_LIMIT)
                                                            + "M.");
                                                    gzFile.delete();
                                                    LOG.info("Local cache file deleted and will not use cache.");
                                                    isCache = false;
                                                }
                                            }
                                        }
                                    } catch (EOFException e) {
                                        finished = true;
                                        e.printStackTrace();
                                    }
                                }
                                reader.close();
                                LOG.info("split " + (i + 1) + " is finished.");
                            }
                            LOG.info("Epoch " + (j + 1) + " finished.");
                            if (isCache) {
                                break;
                            }
                        }
                        osw.close();
                        gos.close();
                    } catch (Exception e) {
                        LOG.warn("Exception in thread stdinRedirectThread");
                        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