A simple solution would be to: 1 - create a constructor of HadoopOutputFormatBase and HadoopOutputFormat that gets the OutputCommitter as a parameter 2 - change the outputCommitter field of HadoopOutputFormatBase to be a generic OutputCommitter 3 - remove the default assignment in the open() and finalizeGlobal to the outputCommitter to FileOutputCommitter(), or keep it as a default in case of no specific assignment.
saluti, Stefano 2015-07-22 16:48 GMT+02:00 Stefano Bortoli <s.bort...@gmail.com>: > In fact, on close() of the HadoopOutputFormat the fileOutputCommitter > returns false on if > (this.fileOutputCommitter.needsTaskCommit(this.context)) returns false. > > i /** > * commit the task by moving the output file out from the temporary > directory. > * @throws java.io.IOException > */ > @Override > public void close() throws IOException { > this.recordWriter.close(new HadoopDummyReporter()); > > if (this.fileOutputCommitter.needsTaskCommit(this.context)) { > this.fileOutputCommitter.commitTask(this.context); > } > } > > > Also, both the close and the finalize global use a FileOutputCommitter, > and never the MongoOutputCommitter > > @Override > public void finalizeGlobal(int parallelism) throws IOException { > > try { > JobContext jobContext = > HadoopUtils.instantiateJobContext(this.jobConf, new JobID()); > FileOutputCommitter fileOutputCommitter = new > FileOutputCommitter(); > > // finalize HDFS output format > fileOutputCommitter.commitJob(jobContext); > } catch (Exception e) { > throw new RuntimeException(e); > } > } > > anyone can have a look into that? > > saluti, > Stefano > > 2015-07-22 15:53 GMT+02:00 Stefano Bortoli <bort...@okkam.it>: > >> Debugging, it seem the commitTask method of the MongoOutputCommitter is >> never called. Is it possible that this 'bulk' approach of mongo-hadoop 1.4 >> does not fit the task execution method of Flink? >> >> any idea? thanks a lot in advance. >> >> saluti, >> Stefano >> >> Stefano Bortoli, PhD >> >> *ENS Technical Director *_______________________________________________ >> *OKKAM**Srl **- www.okkam.it <http://www.okkam.it/>* >> >> *Email:* bort...@okkam.it >> >> *Phone nr: +39 0461 1823912 <%2B39%200461%201823912> * >> >> *Headquarters:* Trento (Italy), Via Trener 8 >> *Registered office:* Trento (Italy), via Segantini 23 >> >> Confidentially notice. This e-mail transmission may contain legally >> privileged and/or confidential information. Please do not read it if you >> are not the intended recipient(S). Any use, distribution, reproduction or >> disclosure by any other person is strictly prohibited. If you have received >> this e-mail in error, please notify the sender and destroy the original >> transmission and its attachments without reading or saving it in any manner. >> >> 2015-07-22 14:26 GMT+02:00 Stefano Bortoli <s.bort...@gmail.com>: >> >>> Hi, >>> >>> I am trying to analyze and update a MongoDB collection with Apache Flink >>> 0.9.0 and Mongo Hadoop 1.4.0 Hadoop 2.6.0. >>> >>> The process is fairly simple, and the MongoInputFormat works smoothly, >>> however it does not write back to the collection. The process works, >>> because the writeAsText works as expected. I am quite puzzled because >>> debugging I can see it writes in some temporary directory. >>> >>> The mapred.output.uri seem to serve just to output a file named >>> _SUCCESS, and if I do not set it fails with >>> java.lang.IllegalArgumentException: Can not create a Path from a null >>> string >>> at org.apache.hadoop.fs.Path.checkPathArg(Path.java:123) >>> at org.apache.hadoop.fs.Path.<init>(Path.java:135) >>> at >>> org.apache.flink.api.java.hadoop.mapreduce.HadoopOutputFormatBase.open(HadoopOutputFormatBase.java:108) >>> at >>> org.apache.flink.runtime.operators.DataSinkTask.invoke(DataSinkTask.java:186) >>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559) >>> at java.lang.Thread.run(Thread.java:745) >>> >>> Anyone experienced something similar? any hints where to look at? Thanks >>> a lot in advance! >>> >>> saluti, >>> Stefano >>> >>> ==================================================== >>> Configuration conf = new Configuration(); >>> conf.set("mapred.output.dir", "/tmp/"); >>> conf.set(MongoUtil.MONGO_INPUT_URI_PROPERTY, >>> collectionsUri); >>> conf.set(MongoUtil.MONGO_OUTPUT_URI_PROPERTY, >>> collectionsUri); >>> >>> Job job = Job.getInstance(conf); >>> >>> // create a MongodbInputFormat, using a Hadoop input format >>> wrapper >>> InputFormat<Object, BSONObject> mapreduceInputFormat = new >>> MyMongoInputFormat<Object, BSONObject>(); >>> HadoopInputFormat<Object, BSONObject> hdIf = new >>> HadoopInputFormat<Object, BSONObject>( >>> mapreduceInputFormat, Object.class, BSONObject.class, >>> job); >>> DataSet<Tuple2<Text, BSONWritable>> fin = input >>> .flatMap(new myFlatMapFunction()).setParallelism(16); >>> >>> MongoConfigUtil.setOutputURI(job.getConfiguration(), >>> collectionsUri); >>> >>> fin.output(new HadoopOutputFormat<Text, BSONWritable>( >>> new MongoOutputFormat<Text, BSONWritable>(), >>> job)); >>> // fin.writeAsText("/tmp/out", WriteMode.OVERWRITE); >>> >>> >> >