Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
MIME-Version: 1.0
In-Reply-To: <CAPb1S7Rs=txmvxhYdTGYqWeaWWpKxsOjO5ZH3xXA6J09Wdh8fQ@mail.gmail.com>
References: <CAPb1S7Rs=txmvxhYdTGYqWeaWWpKxsOjO5ZH3xXA6J09Wdh8fQ@mail.gmail.com>
From: Fabian Hueske <fhueske@gmail.com>
Date: Fri, 20 Oct 2017 23:28:41 +0200
Message-ID: <CAAdrtT32yx2a5RSE02nXzgCz2SQqxFk6ow+sTuEtLpN5hDacaQ@mail.gmail.com>
Subject: Re: problem with increase job parallelism
To: Lei Chen <leyncl2@gmail.com>
Cc: user <user@flink.apache.org>
Content-Type: multipart/alternative; boundary="001a11458462868b88055c012a89"
archived-at: Fri, 20 Oct 2017 21:29:27 -0000

--001a11458462868b88055c012a89
Content-Type: text/plain; charset="UTF-8"

Hi Lei,

setting explicit operator ID should solve this issue.
As far as I know, the auto-generated operator id also depended on the
operator parallelism in previous versions of Flink (not sure until which
point).

Which version are you running?

Best, Fabian


2017-10-17 3:15 GMT+02:00 Lei Chen <leyncl2@gmail.com>:

> Hi,
>
> We're trying to implement some module to help autoscale our pipeline which
> is built  with Flink on YARN. According to the document, the suggested
> procedure seems to be:
>
> 1. cancel job with savepoint
> 2. start new job with increased YARN TM number and parallelism.
>
> However, step 2 always gave error
>
> Caused by: java.lang.IllegalStateException: Failed to rollback to
> savepoint hdfs://10.106.238.14:/tmp/savepoint-767421-20907d234655. Cannot
> map savepoint state for operator 37dfe905df17858e07858039ce3d8ae1 to the
> new program, because the operator is not available in the new program. If
> you want to allow to skip this, you can set the --allowNonRestoredState
> option on the CLI.
> at org.apache.flink.runtime.checkpoint.savepoint.SavepointLoade
> r.loadAndValidateSavepoint(SavepointLoader.java:130)
> at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.re
> storeSavepoint(CheckpointCoordinator.java:1140)
> at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$org$
> apache$flink$runtime$jobmanager$JobManager$$submitJob$1.
> apply$mcV$sp(JobManager.scala:1386)
> at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$org$
> apache$flink$runtime$jobmanager$JobManager$$submitJob$1.
> apply(JobManager.scala:1372)
> at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$org$
> apache$flink$runtime$jobmanager$JobManager$$submitJob$1.
> apply(JobManager.scala:1372)
> at scala.concurrent.impl.Future$PromiseCompletingRunnable.lifte
> dTree1$1(Future.scala:24)
> at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(F
> uture.scala:24)
> at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40)
> at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.
> exec(AbstractDispatcher.scala:397)
> at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
> at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(For
> kJoinPool.java:1339)
> at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPoo
> l.java:1979)
> at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinW
> orkerThread.java:107)
>
> The procedure worked fine if parallelism was not changed.
>
> Also want to mention that I didn't manually specify OperatorID in my job. The
> document does mentioned manually OperatorID assignment is suggested, just
> curious is that mandatory in my case to fix the problem I'm seeing, given
> that my program doesn't change at all so the autogenerated operatorID
> should be unchanged after parallelism increase?
>
> thanks,
> Lei
>

--001a11458462868b88055c012a89
Content-Type: text/html; charset="UTF-8"
Content-Transfer-Encoding: quoted-printable

<div dir=3D"ltr"><div><div><div>Hi Lei,<br></div><div><br></div><div>settin=
g explicit operator ID should solve this issue.<br></div>As far as I know, =
the auto-generated operator id also depended on the operator parallelism in=
 previous versions of Flink (not sure until which point).<br></div><br></di=
v><div>Which version are you running?<br><br></div>Best, Fabian<br><div><br=
></div></div><div class=3D"gmail_extra"><br><div class=3D"gmail_quote">2017=
-10-17 3:15 GMT+02:00 Lei Chen <span dir=3D"ltr">&lt;<a href=3D"mailto:leyn=
cl2@gmail.com" target=3D"_blank">leyncl2@gmail.com</a>&gt;</span>:<br><bloc=
kquote class=3D"gmail_quote" style=3D"margin:0 0 0 .8ex;border-left:1px #cc=
c solid;padding-left:1ex"><div dir=3D"ltr"><span style=3D"font-size:12.8px"=
>Hi,=C2=A0</span><div style=3D"font-size:12.8px"><br></div><div style=3D"fo=
nt-size:12.8px">We&#39;re trying to implement some module to help autoscale=
 our pipeline which is built=C2=A0 with Flink on YARN. According to the doc=
ument, the suggested procedure seems to be:</div><div style=3D"font-size:12=
.8px"><br></div><div style=3D"font-size:12.8px">1. cancel job with savepoin=
t</div><div style=3D"font-size:12.8px">2. start new job with increased YARN=
 TM number and parallelism.=C2=A0</div><div style=3D"font-size:12.8px"><br>=
</div><div style=3D"font-size:12.8px">However, step 2 always gave error=C2=
=A0</div><div style=3D"font-size:12.8px"><br></div><div style=3D"font-size:=
12.8px"><div>Caused by: java.lang.IllegalStateExceptio<wbr>n: Failed to rol=
lback to savepoint hdfs://10.106.238.14:/tmp/save<wbr>point-767421-20907d23=
4655. Cannot map savepoint state for operator 37dfe905df17858e07858039ce3d8=
a<wbr>e1 to the new program, because the operator is not available in the n=
ew program. If you want to allow to skip this, you can set the --allowNonRe=
storedState option on the CLI.</div><div><span style=3D"white-space:pre-wra=
p">	</span>at org.apache.flink.runtime.check<wbr>point.savepoint.SavepointL=
oade<wbr>r.loadAndValidateSavepoint(Sav<wbr>epointLoader.java:130)</div><di=
v><span style=3D"white-space:pre-wrap">	</span>at org.apache.flink.runtime.=
check<a href=3D"http://point.CheckpointCoordinator.re" target=3D"_blank"><w=
br>point.CheckpointCoordinator.re</a><wbr>storeSavepoint(CheckpointCoord<wb=
r>inator.java:1140)</div><div><span style=3D"white-space:pre-wrap">	</span>=
at org.apache.flink.runtime.jobma<wbr>nager.JobManager$$anonfun$org$<wbr>ap=
ache$flink$runtime$jobmanage<wbr>r$JobManager$$submitJob$1.<wbr>apply$mcV$s=
p(JobManager.scala:<wbr>1386)</div><div><span style=3D"white-space:pre-wrap=
">	</span>at org.apache.flink.runtime.jobma<wbr>nager.JobManager$$anonfun$o=
rg$<wbr>apache$flink$runtime$jobmanage<wbr>r$JobManager$$submitJob$1.<wbr>a=
pply(JobManager.scala:1372)</div><div><span style=3D"white-space:pre-wrap">=
	</span>at org.apache.flink.runtime.jobma<wbr>nager.JobManager$$anonfun$org=
$<wbr>apache$flink$runtime$jobmanage<wbr>r$JobManager$$submitJob$1.<wbr>app=
ly(JobManager.scala:1372)</div><div><span style=3D"white-space:pre-wrap">	<=
/span>at scala.concurrent.impl.Future$P<wbr>romiseCompletingRunnable.lifte<=
wbr>dTree1$1(Future.scala:24)</div><div><span style=3D"white-space:pre-wrap=
">	</span>at scala.concurrent.impl.Future$P<wbr>romiseCompletingRunnable.ru=
n(F<wbr>uture.scala:24)</div><div><span style=3D"white-space:pre-wrap">	</s=
pan>at akka.dispatch.TaskInvocation.r<wbr>un(AbstractDispatcher.scala:40<wb=
r>)</div><div><span style=3D"white-space:pre-wrap">	</span>at akka.dispatch=
.ForkJoinExecutor<wbr>Configurator$AkkaForkJoinTask.<wbr>exec(AbstractDispa=
tcher.scala:<wbr>397)</div><div><span style=3D"white-space:pre-wrap">	</spa=
n>at scala.concurrent.forkjoin.Fork<wbr>JoinTask.doExec(ForkJoinTask.j<wbr>=
ava:260)</div><div><span style=3D"white-space:pre-wrap">	</span>at scala.co=
ncurrent.forkjoin.Fork<wbr>JoinPool$WorkQueue.runTask(For<wbr>kJoinPool.jav=
a:1339)</div><div><span style=3D"white-space:pre-wrap">	</span>at scala.con=
current.forkjoin.Fork<wbr>JoinPool.runWorker(ForkJoinPoo<wbr>l.java:1979)</=
div><div><span style=3D"white-space:pre-wrap">	</span>at scala.concurrent.f=
orkjoin.Fork<wbr>JoinWorkerThread.run(ForkJoinW<wbr>orkerThread.java:107)</=
div></div><div style=3D"font-size:12.8px"><br></div><div style=3D"font-size=
:12.8px">The procedure worked fine if parallelism was not changed.=C2=A0</d=
iv><div style=3D"font-size:12.8px"><br></div><div style=3D"font-size:12.8px=
"><span style=3D"font-size:12.8px">Also want to mention that I didn&#39;t m=
anually specify OperatorID in my job.=C2=A0</span>The document does mention=
ed manually OperatorID assignment is suggested, just curious is that mandat=
ory in my case to fix the problem I&#39;m seeing, given that my program doe=
sn&#39;t change at all so the autogenerated operatorID should be unchanged =
after parallelism increase?</div><div style=3D"font-size:12.8px"><br></div>=
<div style=3D"font-size:12.8px">thanks,</div><div style=3D"font-size:12.8px=
">Lei</div></div>
</blockquote></div><br></div>

--001a11458462868b88055c012a89--