[
https://issues.apache.org/jira/browse/FLINK-8712?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16553008#comment-16553008
]
Konstantin Knauf edited comment on FLINK-8712 at 7/23/18 3:38 PM:
------------------------------------------------------------------
{code:java}
Configuration conf = new Configuration();
conf.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 2);
final StreamExecutionEnvironment env =
StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
env.setParallelism(1);
final DataStreamSource<Integer> input = env.addSource(new InfinitySource());
final SingleOutputStreamOperator<Integer> different = input.map(new
MapFunction<Integer, Integer>() {
@Override
public Integer map(Integer integer) throws Exception {
return integer;
}
}).slotSharingGroup("Different");
different.print();
// execute program
env.execute("Flink Streaming Java API Skeleton");
{code}
Works fine for me with Flink 1.5.1. Looking at the Debug logs for above code
snippets it looks like the problem is only that the Taskmanager only offers one
slot, but two are needed obvisously.
Given that we removed the automagical configuration for the YARN CLI shouldnt
we just let the user configure the needed number of slots for the Local
Environment as well?
was (Author: knaufk):
{code:java}
Configuration conf = new Configuration();
conf.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 2);
final StreamExecutionEnvironment env =
StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
env.setParallelism(1);
final DataStreamSource<Integer> input = env.addSource(new InfinitySource());
final SingleOutputStreamOperator<Integer> different = input.map(new
MapFunction<Integer, Integer>() {
@Override
public Integer map(Integer integer) throws Exception {
return integer;
}
}).slotSharingGroup("Different");
different.print();
// execute program
env.execute("Flink Streaming Java API Skeleton");
{code}
Works fine for me with Flink 1.5.1. Looking at the Debug logs for above code
snippets it looks like the problem is only that the Taskmanager only offers one
slot, but two are needed obvisously.
> Cannot execute job with multiple slot sharing groups on LocalExecutor
> ---------------------------------------------------------------------
>
> Key: FLINK-8712
> URL: https://issues.apache.org/jira/browse/FLINK-8712
> Project: Flink
> Issue Type: Bug
> Components: Local Runtime
> Affects Versions: 1.5.0
> Reporter: Till Rohrmann
> Priority: Critical
>
> Currently, it is not possible to run a job with multiple slot sharing groups
> on the LocalExecutor. The problem is that we determine the number of required
> slots simply by looking for the max parallelism of the job but do not
> consider slot sharing groups.
>
> {code:java}
> // set up the streaming execution environment
> final StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> env.setParallelism(1);
> final DataStreamSource<Integer> input = env.addSource(new InfinitySource());
> final SingleOutputStreamOperator<Integer> different = input.map(new
> MapFunction<Integer, Integer>() {
> @Override
> public Integer map(Integer integer) throws Exception {
> return integer;
> }
> }).slotSharingGroup("Different");
> different.print();
> // execute program
> env.execute("Flink Streaming Java API Skeleton");{code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
