Hi Fabian,
that sounds good, thank you.
One final question: As I said earlier, this also distributes data in some
unnecessary cases, say ID 4 sends data to ID 3.
Is there no way to find out the ID of the current node? I guess that number
is already available on the node and just needs to be exposed somehow,
right?
Cheers
Stefan
On 17 September 2015 at 18:39, Fabian Hueske <fhue...@gmail.com> wrote:
> Hi Stefan,
>
> I think I have a solution for your problem :-)
>
> 1) Distribute both parts of the small data to each machine (you have done
> that)
> 2) Your mapper should have a parallelism of 10, the tasks with ID 0 to 4
> (get ID via RichFunction.getRuntimeContext().getIndexOfThisSubtask()) read
> the first half, tasks 5 to 9 read the second half.
> 3) Give the large input into a FlatMapper which sends out two records for
> each incoming record and assigns the first outgoing record a task ID in
> range 0 to 4 and the second outgoing record an ID in range 5 to 9.
> 4) Have a custom partitioner (DataSet.partitionCustom()) after the
> duplicating mapper, which partitions the records based on the assigned task
> Id before they go into the mapper with the other smaller data set. A record
> with assigned task ID 0 will be sent to the mapper task with subtask index
> 0.
>
> This setup is not very nice, but should work.
>
> Let me know, if you need more detail.
>
> Cheers, Fabian
>
> 2015-09-16 21:44 GMT+02:00 Stefan Bunk <stefan.b...@googlemail.com>:
>
>> Hi Fabian,
>>
>> the local file problem would however not exist, if I just copy both
>> halves to all nodes, right?
>>
>> Lets say I have a file `1st` and a file `2nd`, which I copy to all nodes.
>> Now with your approach from above, I do:
>>
>> // helper broadcast datasets to know on which half to operate
>> val data1stHalf = env.fromCollection("1st")
>> val data2ndHalf = env.fromCollection("2nd")
>>
>> val mapped1 = data.flatMap(yourMap).withBroadcastSet(data1stHalf,
>> "fileName").setParallelism(5)
>> val mapped2 = data.flatMap(yourMap).withBroadcastSet(data2ndHalf,
>> "fileName").setParallelism(5)
>> DataSet result = mapped1.union(mapped2)
>>
>> Then, in my custom operator implementation of flatMap I check the helper
>> broadcast data to know which file to load:
>> override def open(params: Configuration): Unit = {
>> val fileName =
>> getRuntimeContext.getBroadcastVariable[String]("fileName")(0)
>> // read the file from the local filesystem which I copied there earlier
>> this.data = loadFromFileIntoDatastructure("/home/data/" + fileName)
>> }
>> override def flatMap(document: Input, out: Collector[Output]): Unit = {
>> // do sth. with this.data and the input
>> out.collect(this.data.process(input))
>> }
>>
>> I think this should work, or do you see another problem here?
>>
>> Which brings us to the other question:
>> The both halves are so large, that one half of the data fits in the
>> user-remaining memory on a node, but not both halves. So my program would
>> probably memory-crash, if the scheduling trusts one node so much, that it
>> wants to execute two flatMaps there ;-).
>>
>> You are saying, that it is not guaranteed, that all 10 nodes are used,
>> but how likely is it, that one node is given two flatMaps and another one
>> is basically idling? I have no idea of the internals, but I guess there is
>> some heuristic inside which decides how to distribute.In the normal setup
>> that all 10 nodes are up, connection is good, all nodes have the same
>> resources available, input data is evenly distributed in HDFS, then the
>> default case should be to distribute to all 10 nodes, right?
>>
>> I am not running in production, so for me it would be ok, if this works
>> out usually.
>>
>> Cheers
>> Stefan
>>
>>
>> On 15 September 2015 at 23:40, Fabian Hueske <fhue...@gmail.com> wrote:
>>
>>> Hi Stefan,
>>>
>>> the problem is that you cannot directly influence the scheduling of
>>> tasks to nodes to ensure that you can read the data that you put in the
>>> local filesystems of your nodes. HDFS gives a shared file system which
>>> means that each node can read data from anywhere in the cluster.
>>> I assumed the data is small enough to broadcast because you want to keep
>>> it in memory.
>>>
>>> Regarding your question. It is not guaranteed that two different tasks,
>>> each with parallelism 5, will be distributed to all 10 nodes (even if you
>>> have only 10 processing slots).
>>> What would work is to have one map task with parallelism 10 and a Flink
>>> setup with 10 task managers on 10 machines with only one processing slot
>>> per TM. However, you won't be able to replicate the data to both sets of
>>> maps because you cannot know which task instance will be executed on which
>>> machine (you cannot distinguish the tasks of both task sets).
>>>
>>> As I said, reading from local file system in a cluster and forcing task
>>> scheduling to specific nodes is quite tricky.
>>> Cheers, Fabian
>>>
>>> 2015-09-15 23:15 GMT+02:00 Stefan Bunk <stefan.b...@googlemail.com>:
>>>
>>>> Hi Fabian,
>>>>
>>>> I think we might have a misunderstanding here. I have already copied
>>>> the first file to five nodes, and the second file to five other nodes,
>>>> outside of Flink. In the open() method of the operator, I just read that
>>>> file via normal Java means. I do not see, why this is tricky or how HDFS
>>>> should help here.
>>>> Then, I have a normal Flink DataSet, which I want to run through the
>>>> operator (using the previously read data in the flatMap implementation). As
>>>> I run the program several times, I do not want to broadcast the data every
>>>> time, but rather just copy it on the nodes, and be fine with it.
>>>>
>>>> Can you answer my question from above? If the setParallelism-method
>>>> works and selects five nodes for the first flatMap and five _other_ nodes
>>>> for the second flatMap, then that would be fine for me if there is no other
>>>> easy solution.
>>>>
>>>> Thanks for your help!
>>>> Best
>>>> Stefan
>>>>
>>>>
>>>> On 14 September 2015 at 22:28, Fabian Hueske <fhue...@gmail.com> wrote:
>>>>
>>>>> Hi Stefan,
>>>>>
>>>>> forcing the scheduling of tasks to certain nodes and reading files
>>>>> from the local file system in a multi-node setup is actually quite tricky
>>>>> and requires a bit understanding of the internals.
>>>>> It is possible and I can help you with that, but would recommend to
>>>>> use a shared filesystem such as HDFS if that is possible.
>>>>>
>>>>> Best, Fabian
>>>>>
>>>>> 2015-09-14 19:16 GMT+02:00 Stefan Bunk <stefan.b...@googlemail.com>:
>>>>>
>>>>>> Hi,
>>>>>>
>>>>>> actually, I am distributing my data before the program starts,
>>>>>> without using broadcast sets.
>>>>>>
>>>>>> However, the approach should still work, under one condition:
>>>>>>
>>>>>>> DataSet mapped1 =
>>>>>>> data.flatMap(yourMap).withBroadcastSet(smallData1,"data").setParallelism(5);
>>>>>>> DataSet mapped2 =
>>>>>>> data.flatMap(yourMap).withBroadcastSet(smallData2,"data").setParallelism(5);
>>>>>>>
>>>>>> Is it guaranteed, that this selects a disjoint set of nodes, i.e.
>>>>>> five nodes for mapped1 and five other nodes for mapped2?
>>>>>>
>>>>>> Is there any way of selecting the five nodes concretely? Currently, I
>>>>>> have stored the first half of the data on nodes 1-5 and the second half
>>>>>> on
>>>>>> nodes 6-10. With this approach, I guess, nodes are selected randomly so I
>>>>>> would have to copy both halves to all of the nodes.
>>>>>>
>>>>>> Best,
>>>>>> Stefan
>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>>
>
