Which version of Flink are you using? On Tue, Mar 5, 2019 at 10:58 PM Le Xu <sharonx...@gmail.com> wrote:
> Hi Till: > > Thanks for the reply. The setup of the jobs is roughly as follows: For a > cluster with N machines, we deploy X simple map/reduce style jobs (the job > DAG and settings are exactly the same, except they consumes different > data). Each job has N mappers (they are evenly distributed, one mapper on > each machine).There are X mappers on each machine (as there are X jobs in > total). Each job has only one reducer where all mappers point to. What I'm > observing is that all reducers are allocated to machine 1 (where all mapper > 1 from every job is allocated to). It does make sense since reducer and > mapper 1 are in the same slot group. The original purpose of the questions > is to find out whether it is possible to explicitly specify that reducer > can be co-located with another mapper (such as mapper 2 so the reducer of > job 2 can be placed on machine 2). Just trying to figure out if it is all > possible without using more expensive approach (through YARN for example). > But if it is not possible I will see if I can move to job mode as Piotr > suggests. > > Thanks, > > Le > > On Tue, Mar 5, 2019 at 9:24 AM Till Rohrmann <trohrm...@apache.org> wrote: > >> Hard to tell whether this is related to FLINK-11815. >> >> To me the setup is not fully clear. Let me try to sum it up: According to >> Le Xu's description there are n jobs running on a session cluster. I assume >> that every TaskManager has n slots. The observed behaviour is that every >> job allocates the slot for the first mapper and chained sink from the first >> TM, right? Since Flink does not give strict guarantees for the slot >> allocation this is possible, however it should be highly unlikely or at >> least change when re-executing the same setup. At the moment there is no >> functionality in place to control the task-slot assignment. >> >> Chaining only affects which task will be grouped together and executed by >> the same Task (being executed by the same thread). Separate tasks can still >> be executed in the same slot if they have the same slot sharing group. This >> means that there can be multiple threads running in each slot. >> >> For me it would be helpful to get more information about the actual job >> deployments. >> >> Cheers, >> Till >> >> On Tue, Mar 5, 2019 at 12:00 PM Piotr Nowojski <pi...@ververica.com> >> wrote: >> >>> Hi Le, >>> >>> As I wrote, you can try running Flink in job mode, which spawns separate >>> clusters per each job. >>> >>> Till, is this issue covered by FLINK-11815 >>> <https://issues.apache.org/jira/browse/FLINK-11815> ? Is this the same >>> as: >>> >>> > Known issues: >>> > 1. (…) >>> > 2. if task slots are registered before slot request, the code have a >>> tendency to group requests together on the same machine because we >>> are using a LinkedHashMap >>> >>> ? >>> >>> Piotrek >>> >>> On 4 Mar 2019, at 21:08, Le Xu <sharonx...@gmail.com> wrote: >>> >>> Thanks Piotr. >>> >>> I didn't realize that the email attachment isn't working so the example >>> I was referring to was this figure from Flink website: >>> https://ci.apache.org/projects/flink/flink-docs-stable/fig/slot_sharing.svg >>> >>> So I try to run multiple jobs concurrently in a cluster -- the jobs are >>> identical and the DAG looks very similar to the one in the figure. Each >>> machine holds one map task from each job. I end up with X number of sinks >>> on machine 1 (X being the number of jobs). I assume this is caused by the >>> operator chaining (so that all sinks are chained to mapper 1 all end up on >>> machine 1). But I also tried disabling chaining but I still get the same >>> result. Some how even when the sink and the map belongs to different >>> threads they are still placed in the same slot. >>> >>> My goal was to see whether it is possible to have sinks evenly >>> distributed across the cluster (instead of all on machine 1). One way to do >>> this is to see if it is ok to chained the sink to one of the other mapper >>> -- the other way is to see if we can change the placement of the mapper >>> altogether (like placing map 1 of job 2 on machine 2, map 1 of job 3 on >>> machine 3 so we end up with sinks sit evenly throughout the cluster). >>> >>> Thanks. >>> >>> Le >>> >>> On Mon, Mar 4, 2019 at 6:49 AM Piotr Nowojski <pi...@ververica.com> >>> wrote: >>> >>>> Hi, >>>> >>>> Are you asking the question if that’s the behaviour or you have >>>> actually observed this issue? I’m not entirely sure, but I would guess that >>>> the Sink tasks would be distributed randomly across the cluster, but maybe >>>> I’m mixing this issue with resource allocations for Task Managers. Maybe >>>> Till will know something more about this? >>>> >>>> One thing that might have solve/workaround the issue is to run those >>>> jobs in the job mode (one cluster per job), not in cluster mode, since >>>> containers for Task Managers are created/requested randomly. >>>> >>>> Piotrek >>>> >>>> On 2 Mar 2019, at 23:53, Le Xu <sharonx...@gmail.com> wrote: >>>> >>>> Hello! >>>> >>>> I'm trying to find out if there a way to force task slot sharing within >>>> a job. The example on the website looks like the following (as in the >>>> screenshot) >>>> >>>> <image.png> >>>> In this example, the single sink is slot-sharing with source/map (1) >>>> and window operator (1). If I deploy multiple identical jobs shown above, >>>> all sink operators would be placed on the first machine (which creates an >>>> unbalanced scenario). Is there a way to avoid this situation (i.e., to have >>>> sink operators of different jobs spread evenly across the task slots for >>>> the entire cluster). Specifically, I was wondering if either of the >>>> following options are possible: >>>> 1. To force Sink[1] to be slot sharing with mapper from a different >>>> partition on other slots such as (source[2] and window[2]). >>>> 2. If option 1 is not possible, is there a "hacky" way for Flink to >>>> deploy jobs starting from a different machine: e.g. For job 2, it can >>>> allocate source/map[1], window[1], sink[1] to machine 2 instead of again on >>>> machine 1. In this way the slot-sharing groups are still the same, but we >>>> end up having sinks from the two jobs on different machines. >>>> >>>> >>>> Thanks! >>>> >>>> >>>> >>>> >>>> >>>> >>>
