You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Will Walters <ww...@yahoo-inc.com> on 2017/06/06 00:13:37 UTC
Issue With Configuration File
Hello,
I'm having issues editing the default Flink memory settings. I'm attempting to run a Flink task on a cluster at scale. The log shows my edited config settings having been read into the program, but they're having no effect. Here's the trace:
17/06/05 23:45:41 INFO flink.FlinkRunner: Starting execution of Flink program.17/06/05 23:45:41 INFO java.ExecutionEnvironment: The job has 0 registered types and 0 default Kryo serializers17/06/05 23:45:41 INFO configuration.GlobalConfiguration: Loading configuration property: taskmanager.memory.size, 10017/06/05 23:45:41 INFO configuration.GlobalConfiguration: Loading configuration property: taskmanager.heap.mb, 25617/06/05 23:45:41 INFO configuration.GlobalConfiguration: Loading configuration property: taskmanager.debug.memory.startLogThread, true17/06/05 23:45:41 INFO minicluster.FlinkMiniCluster: Disabled queryable state server17/06/05 23:45:41 INFO minicluster.FlinkMiniCluster: Starting FlinkMiniCluster....17/06/05 23:45:41 INFO network.NetworkEnvironment: Starting the network environment and its components.17/06/05 23:45:41 INFO taskmanager.TaskManager: Limiting managed memory to 17592186044406 MB, memory will be allocated lazily.17/06/05 23:45:41 ERROR flink.FlinkRunner: Pipeline execution failedjava.lang.IllegalArgumentException: Size of total memory must be positive.
And here's the config file I'm using:
taskmanager.memory.size: 100taskmanager.heap.mb: 256taskmanager.debug.memory.startLogThread: true
In the shell script I'm using to run the task, I've edited the FLINK_CONF_DIR to direct to the config file that I created.
If anyone has any advice or inputs it would be much appreciated.
Thanks!Will Walters.
Re: Issue With Configuration File
Posted by Aljoscha Krettek <al...@apache.org>.
Hi Will,
I’m afraid that will simply run your program in Flink local mode, where the configuration settings are ignored because an in-process Flink Cluster is being started.
For running a Beam pipeline on YARN you have two options right now:
1. Start a Flink YARN session as described here https://ci.apache.org/projects/flink/flink-docs-release-1.3/setup/yarn_setup.html#flink-yarn-session <https://ci.apache.org/projects/flink/flink-docs-release-1.3/setup/yarn_setup.html#flink-yarn-session> and submit your Beam job as described here https://beam.apache.org/get-started/quickstart-java/#run-wordcount <https://beam.apache.org/get-started/quickstart-java/#run-wordcount> once the YARN session is up and has printed the JobManager details that you need to submit the Beam job.
2. Package your program as a fat jar. The quickstart from the Beam website should have the rights settings in the POM for doing that. Then use regular Flink submission to submit that jar to a YARN cluster as described here https://ci.apache.org/projects/flink/flink-docs-release-1.3/setup/yarn_setup.html#run-a-single-flink-job-on-yarn <https://ci.apache.org/projects/flink/flink-docs-release-1.3/setup/yarn_setup.html#run-a-single-flink-job-on-yarn>
Best,
Aljoscha
> On 6. Jun 2017, at 19:05, Will Walters <ww...@yahoo-inc.com> wrote:
>
> Aljoscha,
>
> You're correct that I'm using Beam. Here's the shell script I'm using to start the job:
>
> hadoop jar **path to jar file** org.apache.beam.examples.complete.TfIdf \
> --runner=FlinkRunner \
> --input= **path to directory** \
> --output=tfIdf
>
> There are two lines before that to set the Flink config path, as well as the Hadoop classpath.
>
> Thank you,
> Will.
>
>
> On Tuesday, June 6, 2017 5:45 AM, Aljoscha Krettek <al...@apache.org> wrote:
>
>
> Hi Will,
>
> How are you starting your cluster/executing your program? From what I can gather you are using Beam, is that right? The line about the FlinkMiniCluster seems strange because this would hint at the fact that the Runner is trying to execute in local mode.
>
> Best,
> Aljoscha
>
>> On 6. Jun 2017, at 02:13, Will Walters <wwalters@yahoo-inc.com <ma...@yahoo-inc.com>> wrote:
>>
>> Hello,
>>
>> I'm having issues editing the default Flink memory settings. I'm attempting to run a Flink task on a cluster at scale. The log shows my edited config settings having been read into the program, but they're having no effect. Here's the trace:
>>
>> 17/06/05 23:45:41 INFO flink.FlinkRunner: Starting execution of Flink program.
>> 17/06/05 23:45:41 INFO java.ExecutionEnvironment: The job has 0 registered types and 0 default Kryo serializers
>> 17/06/05 23:45:41 INFO configuration.GlobalConfiguration: Loading configuration property: taskmanager.memory.size, 100
>> 17/06/05 23:45:41 INFO configuration.GlobalConfiguration: Loading configuration property: taskmanager.heap.mb, 256
>> 17/06/05 23:45:41 INFO configuration.GlobalConfiguration: Loading configuration property: taskmanager.debug.memory.startLogThread, true
>> 17/06/05 23:45:41 INFO minicluster.FlinkMiniCluster: Disabled queryable state server
>> 17/06/05 23:45:41 INFO minicluster.FlinkMiniCluster: Starting FlinkMiniCluster.
>> .
>> .
>> .
>> 17/06/05 23:45:41 INFO network.NetworkEnvironment: Starting the network environment and its components.
>> 17/06/05 23:45:41 INFO taskmanager.TaskManager: Limiting managed memory to 17592186044406 MB, memory will be allocated lazily.
>> 17/06/05 23:45:41 ERROR flink.FlinkRunner: Pipeline execution failed
>> java.lang.IllegalArgumentException: Size of total memory must be positive.
>>
>> And here's the config file I'm using:
>>
>> taskmanager.memory.size: 100
>> taskmanager.heap.mb: 256
>> taskmanager.debug.memory.startLogThread: true
>>
>> In the shell script I'm using to run the task, I've edited the FLINK_CONF_DIR to direct to the config file that I created.
>>
>> If anyone has any advice or inputs it would be much appreciated.
>>
>> Thanks!
>> Will Walters.
>
>
>
Re: Issue With Configuration File
Posted by Will Walters <ww...@yahoo-inc.com>.
Aljoscha,
You're correct that I'm using Beam. Here's the shell script I'm using to start the job:
hadoop jar **path to jar file** org.apache.beam.examples.complete.TfIdf \--runner=FlinkRunner \--input= **path to directory** \--output=tfIdf
There are two lines before that to set the Flink config path, as well as the Hadoop classpath.
Thank you,Will.
On Tuesday, June 6, 2017 5:45 AM, Aljoscha Krettek <al...@apache.org> wrote:
Hi Will,
How are you starting your cluster/executing your program? From what I can gather you are using Beam, is that right? The line about the FlinkMiniCluster seems strange because this would hint at the fact that the Runner is trying to execute in local mode.
Best,Aljoscha
On 6. Jun 2017, at 02:13, Will Walters <ww...@yahoo-inc.com> wrote:
Hello,
I'm having issues editing the default Flink memory settings. I'm attempting to run a Flink task on a cluster at scale. The log shows my edited config settings having been read into the program, but they're having no effect. Here's the trace:
17/06/05 23:45:41 INFO flink.FlinkRunner: Starting execution of Flink program.17/06/05 23:45:41 INFO java.ExecutionEnvironment: The job has 0 registered types and 0 default Kryo serializers17/06/05 23:45:41 INFO configuration.GlobalConfiguration: Loading configuration property: taskmanager.memory.size, 10017/06/05 23:45:41 INFO configuration.GlobalConfiguration: Loading configuration property: taskmanager.heap.mb, 25617/06/05 23:45:41 INFO configuration.GlobalConfiguration: Loading configuration property: taskmanager.debug.memory.startLogThread, true17/06/05 23:45:41 INFO minicluster.FlinkMiniCluster: Disabled queryable state server17/06/05 23:45:41 INFO minicluster.FlinkMiniCluster: Starting FlinkMiniCluster....17/06/05 23:45:41 INFO network.NetworkEnvironment: Starting the network environment and its components.17/06/05 23:45:41 INFO taskmanager.TaskManager: Limiting managed memory to 17592186044406 MB, memory will be allocated lazily.17/06/05 23:45:41 ERROR flink.FlinkRunner: Pipeline execution failedjava.lang.IllegalArgumentException: Size of total memory must be positive.
And here's the config file I'm using:
taskmanager.memory.size: 100taskmanager.heap.mb: 256taskmanager.debug.memory.startLogThread: true
In the shell script I'm using to run the task, I've edited the FLINK_CONF_DIR to direct to the config file that I created.
If anyone has any advice or inputs it would be much appreciated.
Thanks!Will Walters.
Re: Issue With Configuration File
Posted by Aljoscha Krettek <al...@apache.org>.
Hi Will,
How are you starting your cluster/executing your program? From what I can gather you are using Beam, is that right? The line about the FlinkMiniCluster seems strange because this would hint at the fact that the Runner is trying to execute in local mode.
Best,
Aljoscha
> On 6. Jun 2017, at 02:13, Will Walters <ww...@yahoo-inc.com> wrote:
>
> Hello,
>
> I'm having issues editing the default Flink memory settings. I'm attempting to run a Flink task on a cluster at scale. The log shows my edited config settings having been read into the program, but they're having no effect. Here's the trace:
>
> 17/06/05 23:45:41 INFO flink.FlinkRunner: Starting execution of Flink program.
> 17/06/05 23:45:41 INFO java.ExecutionEnvironment: The job has 0 registered types and 0 default Kryo serializers
> 17/06/05 23:45:41 INFO configuration.GlobalConfiguration: Loading configuration property: taskmanager.memory.size, 100
> 17/06/05 23:45:41 INFO configuration.GlobalConfiguration: Loading configuration property: taskmanager.heap.mb, 256
> 17/06/05 23:45:41 INFO configuration.GlobalConfiguration: Loading configuration property: taskmanager.debug.memory.startLogThread, true
> 17/06/05 23:45:41 INFO minicluster.FlinkMiniCluster: Disabled queryable state server
> 17/06/05 23:45:41 INFO minicluster.FlinkMiniCluster: Starting FlinkMiniCluster.
> .
> .
> .
> 17/06/05 23:45:41 INFO network.NetworkEnvironment: Starting the network environment and its components.
> 17/06/05 23:45:41 INFO taskmanager.TaskManager: Limiting managed memory to 17592186044406 MB, memory will be allocated lazily.
> 17/06/05 23:45:41 ERROR flink.FlinkRunner: Pipeline execution failed
> java.lang.IllegalArgumentException: Size of total memory must be positive.
>
> And here's the config file I'm using:
>
> taskmanager.memory.size: 100
> taskmanager.heap.mb: 256
> taskmanager.debug.memory.startLogThread: true
>
> In the shell script I'm using to run the task, I've edited the FLINK_CONF_DIR to direct to the config file that I created.
>
> If anyone has any advice or inputs it would be much appreciated.
>
> Thanks!
> Will Walters.