| 67 | |
| 68 | |
| 69 | public void store(DataBag data, FuncSpec storeFuncSpec, PigContext pigContext) throws IOException { |
| 70 | Configuration conf = ConfigurationUtil.toConfiguration(pigContext.getProperties()); |
| 71 | // create a simulated JobContext |
| 72 | JobContext jc = HadoopShims.createJobContext(conf, new JobID()); |
| 73 | StoreFuncInterface sfunc = (StoreFuncInterface)PigContext.instantiateFuncFromSpec( |
| 74 | storeFuncSpec); |
| 75 | OutputFormat<?,?> of = sfunc.getOutputFormat(); |
| 76 | |
| 77 | POStore store = new POStore(new OperatorKey()); |
| 78 | store.setSFile(new FileSpec(file, storeFuncSpec)); |
| 79 | PigOutputFormat.setLocation(jc, store); |
| 80 | OutputCommitter oc; |
| 81 | // create a simulated TaskAttemptContext |
| 82 | |
| 83 | TaskAttemptContext tac = HadoopShims.createTaskAttemptContext(conf, HadoopShims.getNewTaskAttemptID()); |
| 84 | PigOutputFormat.setLocation(tac, store); |
| 85 | RecordWriter<?,?> rw ; |
| 86 | try { |
| 87 | of.checkOutputSpecs(jc); |
| 88 | oc = of.getOutputCommitter(tac); |
| 89 | oc.setupJob(jc); |
| 90 | oc.setupTask(tac); |
| 91 | rw = of.getRecordWriter(tac); |
| 92 | sfunc.prepareToWrite(rw); |
| 93 | |
| 94 | for (Iterator<Tuple> it = data.iterator(); it.hasNext();) { |
| 95 | Tuple row = it.next(); |
| 96 | sfunc.putNext(row); |
| 97 | } |
| 98 | rw.close(tac); |
| 99 | } catch (InterruptedException e) { |
| 100 | throw new IOException(e); |
| 101 | } |
| 102 | if(oc.needsTaskCommit(tac)) { |
| 103 | oc.commitTask(tac); |
| 104 | } |
| 105 | oc.commitJob(jc); |
| 106 | } |
| 107 | |
| 108 | @Override |
| 109 | public String toString() { |