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 >>>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>> . >>>>>> >>>>>> >>>>> >>> >> >
