Hi! Can you: - either return a BSONWritable from the function - or type the output formats to String?
The MongoRecordWriter can work with non BSON objects as well. https://github.com/mongodb/mongo-hadoop/blob/master/core/src/main/java/com/mongodb/hadoop/mapred/output/MongoRecordWriter.java Stephan On Thu, Nov 6, 2014 at 4:12 PM, Flavio Pompermaier <[email protected]> wrote: > I'm trying to do that but I can't find the proper typing.. For example: > > DataSet<String> fin = input.map(new MapFunction<Tuple2<BSONWritable, > BSONWritable>, String>() { > > private static final long serialVersionUID = 1L; > > @Override > public String map(Tuple2<BSONWritable, BSONWritable> record) throws > Exception { > BSONWritable value = record.getField(1); > BSONObject doc = value.getDoc(); > BasicDBObject jsonld = (BasicDBObject) doc.get("jsonld"); > String type = jsonld.getString("@type"); > return type; > } > }); > > MongoConfigUtil.setOutputURI( hdIf.getJobConf(), > "mongodb://localhost:27017/test.test"); > fin.output(new HadoopOutputFormat<BSONWritable,BSONWritable>(new > MongoOutputFormat<BSONWritable,BSONWritable>(), hdIf.getJobConf())); > > Obviously this doesn't work because I'm emitting strings and trying to > write BSONWritable ..can you show me a simple working example? > > Best, > Flavio > > On Thu, Nov 6, 2014 at 3:58 PM, Stephan Ewen <[email protected]> wrote: > >> Hi Flavio! >> >> I think the general method is the same as with the inputs. >> >> You use the "HadoopOutputFormat" wrapping the "MongoOutputFormat" ( >> https://github.com/mongodb/mongo-hadoop/blob/master/core/src/main/java/com/mongodb/hadoop/mapred/MongoOutputFormat.java >> ) >> >> You can then call >> >> DataSet<Tuple2<BSONWritable, BSONWritable>> data = ...; >> >> data.output(mongoOutput); >> >> >> Greetings, >> Stephan >> >> >> On Thu, Nov 6, 2014 at 3:41 PM, Flavio Pompermaier <[email protected]> >> wrote: >> >>> Any help here..? >>> >>> On Wed, Nov 5, 2014 at 6:39 PM, Flavio Pompermaier <[email protected] >>> > wrote: >>> >>>> Just shared the example at >>>> https://github.com/okkam-it/flink-mongodb-test and twitted :) >>>> >>>> The next step is to show how to write the result of a Flink process >>>> back to Mongo. >>>> How can I manage to do that? Can someone help me? >>>> >>>> On Wed, Nov 5, 2014 at 1:17 PM, Fabian Hueske <[email protected]> >>>> wrote: >>>> >>>>> How about going for an optional parameter for the InputFormat to >>>>> determine into how many splits each region is split? >>>>> That would be a lightweight option to control the number of splits >>>>> with low effort (on our side). >>>>> >>>>> 2014-11-05 0:01 GMT+01:00 Flavio Pompermaier <[email protected]>: >>>>> >>>>>> So how are we going to proceed here? Is someone willing to help me in >>>>>> improving the splitting policy or we leave it as it is now? >>>>>> >>>>>> >>>>>> On Tue, Nov 4, 2014 at 9:42 PM, Fabian Hueske <[email protected]> >>>>>> wrote: >>>>>> >>>>>>> I agree. Going for more splits with smaller key regions is a >>>>>>> good idea. >>>>>>> However, it might be a bit difficult to determine a good number of >>>>>>> splits as the size of a split depends on its density. Too large splits >>>>>>> are >>>>>>> prone to cause data skew, too small ones will increase the overhead of >>>>>>> split assignment. >>>>>>> >>>>>>> A solution for this problem could be to add an optional parameter to >>>>>>> the IF to give an upper bound for the number of InputSplits. >>>>>>> >>>>>>> 2014-11-04 20:53 GMT+01:00 Stephan Ewen <[email protected]>: >>>>>>> >>>>>>>> Typo: it should have meant that workers that get a larger split >>>>>>>> will get fewer additional splits. >>>>>>>> Am 04.11.2014 20:48 schrieb [email protected]: >>>>>>>> >>>>>>>> InputSplits are assigned lazily at runtime, which gives you many of >>>>>>>> the benefits of re-assigning without the nastyness. >>>>>>>> >>>>>>>> Can you write the logic that creates the splits such that it >>>>>>>> creates multiple splits per region? Then the lazy assignment will make >>>>>>>> sure >>>>>>>> that workers that get a larger split will get get additional splits >>>>>>>> than >>>>>>>> workers that get smaller splits... >>>>>>>> Am 04.11.2014 20:32 schrieb "Fabian Hueske" <[email protected]>: >>>>>>>> >>>>>>>> Hmm, that's good question indeed. I am not familiar with HBase's >>>>>>>>> mode of operation. >>>>>>>>> I would assume, that HBase uses range partitioning to partition a >>>>>>>>> table into regions. That way it is rather easy to balance the size of >>>>>>>>> regions, as long as there is no single key that occurs very often. I >>>>>>>>> am not >>>>>>>>> sure if it is possible to overcome data skew cause by frequent keys. >>>>>>>>> However as I said, these are just assumption. I will have a look >>>>>>>>> at HBase's internals for verification. >>>>>>>>> >>>>>>>>> In any case, Flink does currently not support reassigning >>>>>>>>> or splitting of InputSplits at runtime. >>>>>>>>> Also initially generating balanced InputSplits willl be >>>>>>>>> tricky. That would be possible if we can efficiently determine the >>>>>>>>> "density" of a key range when creating the InputSplits. However, I'm >>>>>>>>> a bit >>>>>>>>> skeptical that this can be done... >>>>>>>>> >>>>>>>>> 2014-11-04 17:33 GMT+01:00 Flavio Pompermaier < >>>>>>>>> [email protected]>: >>>>>>>>> >>>>>>>>>> From what I know HBase manages the regions but the fact that they >>>>>>>>>> are evenly distributed depends on a well-designed key.. >>>>>>>>>> if it is not the case you could encounter very unbalanced regions >>>>>>>>>> (i.e. hot spotting). >>>>>>>>>> >>>>>>>>>> Could it be a good idea to create a split policy that compares >>>>>>>>>> the size of all the splits and generate equally-sized split that can >>>>>>>>>> be >>>>>>>>>> reassigned to free worker if the original assigned one is still busy? >>>>>>>>>> >>>>>>>>>> On Tue, Nov 4, 2014 at 5:18 PM, Fabian Hueske <[email protected] >>>>>>>>>> > wrote: >>>>>>>>>> >>>>>>>>>>> ad 1) HBase manages the regions and should also take care >>>>>>>>>>> of their uniform size. >>>>>>>>>>> as 2) Dynamically changing InputSplits is not possible at the >>>>>>>>>>> moment. However, the input split generation of the IF should also >>>>>>>>>>> be able >>>>>>>>>>> to handle such issues upfront. In fact, the IF could also generate >>>>>>>>>>> multiple >>>>>>>>>>> splits per region (this would be necessary to make sure that the >>>>>>>>>>> minimum >>>>>>>>>>> number of splits is generated if there are less regions than >>>>>>>>>>> required >>>>>>>>>>> splits). >>>>>>>>>>> >>>>>>>>>>> 2014-11-04 17:04 GMT+01:00 Flavio Pompermaier < >>>>>>>>>>> [email protected]>: >>>>>>>>>>> >>>>>>>>>>>> Ok, thanks for the explanation! >>>>>>>>>>>> That was more or less like I thought it should be but there are >>>>>>>>>>>> still points I'd like to clarify: >>>>>>>>>>>> >>>>>>>>>>>> 1 - What if a region is very big and there are other regions >>>>>>>>>>>> very small..? There will be one slot that takes a very long time >>>>>>>>>>>> while the >>>>>>>>>>>> others will stay inactive.. >>>>>>>>>>>> 2 - Do you think it is possible to implement this in an >>>>>>>>>>>> adaptive way (stop processing of huge region if it worth it and >>>>>>>>>>>> assign >>>>>>>>>>>> remaining data to inactive task managers)? >>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>>> On Tue, Nov 4, 2014 at 4:37 PM, Fabian Hueske < >>>>>>>>>>>> [email protected]> wrote: >>>>>>>>>>>> >>>>>>>>>>>>> Local split assignment preferably assigns input split to >>>>>>>>>>>>> workers that can locally read the data of an input split. >>>>>>>>>>>>> For example, HDFS stores file chunks (blocks) distributed over >>>>>>>>>>>>> the cluster and gives access to these chunks to every worker via >>>>>>>>>>>>> network >>>>>>>>>>>>> transfer. However, if a chunk is read from a process that runs on >>>>>>>>>>>>> the same >>>>>>>>>>>>> node as the chunk is stored, the read operation directly accesses >>>>>>>>>>>>> the local >>>>>>>>>>>>> file system without going over the network. Hence, it is >>>>>>>>>>>>> essential to >>>>>>>>>>>>> assign input splits based on the locality of their data if you >>>>>>>>>>>>> want to have >>>>>>>>>>>>> reasonably performance. We call this local split assignment. This >>>>>>>>>>>>> is a >>>>>>>>>>>>> general concept of all data parallel systems including Hadoop, >>>>>>>>>>>>> Spark, and >>>>>>>>>>>>> Flink. >>>>>>>>>>>>> >>>>>>>>>>>>> This issue is not related to serializability of input formats. >>>>>>>>>>>>> I assume that the wrapped MongoIF is also not capable of local >>>>>>>>>>>>> split assignment. >>>>>>>>>>>>> >>>>>>>>>>>>> Am Dienstag, 4. November 2014 schrieb Flavio Pompermaier : >>>>>>>>>>>>> >>>>>>>>>>>>>> What do you mean for "might lack support for local split >>>>>>>>>>>>>> assignment"? You mean that InputFormat is not serializable? >>>>>>>>>>>>>> This instead is not true for Mongodb? >>>>>>>>>>>>>> >>>>>>>>>>>>>> On Tue, Nov 4, 2014 at 10:00 AM, Fabian Hueske < >>>>>>>>>>>>>> [email protected]> wrote: >>>>>>>>>>>>>> >>>>>>>>>>>>>>> There's a page about Hadoop Compatibility that shows how to >>>>>>>>>>>>>>> use the wrapper. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> The HBase format should work as well, but might lack support >>>>>>>>>>>>>>> for local split assignment. In that case performance would >>>>>>>>>>>>>>> suffer a lot. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Am Dienstag, 4. November 2014 schrieb Flavio Pompermaier : >>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Should I start from >>>>>>>>>>>>>>>> http://flink.incubator.apache.org/docs/0.7-incubating/example_connectors.html >>>>>>>>>>>>>>>> ? Is it ok? >>>>>>>>>>>>>>>> Thus, in principle, also the TableInputFormat of HBase >>>>>>>>>>>>>>>> could be used in a similar way..isn't it? >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> On Tue, Nov 4, 2014 at 9:42 AM, Fabian Hueske < >>>>>>>>>>>>>>>> [email protected]> wrote: >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> Hi, >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> the blog post uses Flinks wrapper for Hadoop InputFormats. >>>>>>>>>>>>>>>>> This has been ported to the new API and is described in >>>>>>>>>>>>>>>>> the documentation. >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> So you just need to take Mongos Hadoop IF and plug it into >>>>>>>>>>>>>>>>> the new IF wrapper. :-) >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> Fabian >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> Am Dienstag, 4. November 2014 schrieb Flavio Pompermaier : >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> Hi to all, >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> I saw this post >>>>>>>>>>>>>>>>>> https://flink.incubator.apache.org/news/2014/01/28/querying_mongodb.html >>>>>>>>>>>>>>>>>> but it use the old APIs (HadoopDataSource instead of >>>>>>>>>>>>>>>>>> DataSource). >>>>>>>>>>>>>>>>>> How can I use Mongodb with the new Flink APIs? >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> Best, >>>>>>>>>>>>>>>>>> Flavio >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> . >>>>>>>>>> >>>>>>>>>> >>>>>>>>> >>>>>>> >>>>>> >>>>> >>>> >> >
