Re: Auto/Dynamic scaling in Flink

[罗齐]

Hi Nauroz,

If you’re using Flink 1.5 on Yarn, it supports dynamic task manager allocation 
by default [1]. After skimming the code, it seems to me that in general if 
requested parallelism is larger than available task slots, new task managers 
will be requested via ResourceManager (please correct me if any 
misunderstanding).

We’re also looking into auto scaling issue in Flink, as it’s crucial in our use 
cases (e.g. in batch mode, we will do input with few TMs but output with much 
more TMs). Please kindly let me know if any further information are available.

[1] 
https://stackoverflow.com/questions/38054015/can-yarn-dynamically-allocate-resources-to-flink/38077626#38077626
 


> On Nov 13, 2018, at 8:10 PM, Tzu-Li Chen  wrote:
> 
> Hi,
> 
> Yet Flink does not support auto-scaling. However, there is an umbrella JIRA 
> issue[1]
> to cover the discussion about it. And I think the design doc(draft) 
> attached[2] could help.
> 
> Best,
> tison.
> 
> [1] https://issues.apache.org/jira/browse/FLINK-10407 
> 
> [2] 
> https://docs.google.com/document/d/1XKDXnrp8w45k2jIJNHxpNP2Zmr6FGru3H0dTHwYWJyE/edit
>  
> 
> 
> Tzu-Li (Gordon) Tai mailto:tzuli...@apache.org>> 
> 于2018年11月13日周二 下午8:05写道：
> Hi,
> 
> Flink does not support auto-scaling, yet. Rescaling operations currently are 
> always manual, i.e take a savepoint of the Flink job, and when restoring from 
> the savepoint, define a new parallelism for the job.
> As for the metrics to be used for auto-scaling, I can imagine that it would 
> be possible to base this on top of metrics such as TM throughput, 
> backpressure, etc. AFAIK, there also currently isn't any official design or 
> discussion on going for this.
> 
> Cheers,
> Gordon
> 
> On Mon, Nov 12, 2018 at 5:50 PM Nauroz Khan Nausherwani 
>  > wrote:
> Dear Flink Contributors and users,
> 
> I am a PhD student and I was interested to know, using which matrices, and 
> when does Flink performs scaling-in or scaling out of resources? I did search 
> the flink's website where I could only find information about how dynamic 
> scaling is performed in stateless or stateful operator.  It would be 
> interesting to know which matrices Flink uses, and when actually Flink 
> triggers auto-scaling.
> 
> Any link or reference paper with required information is appreciated.
> 
> best regards,
> Nauroz

Flink Mini Cluster Performance?

[jlist9]

I'm developing a Flink application and I'm still learning. For simplicity,
most of the time I test by running the main method of the entry class as a
regular Java application. Would that be running on what's called a mini
cluster? I find it quote convenient and makes debugging job really easy. My
question is, if it's a job that's small enough and can potentially be
executed on a single machine, is there going to be a performance penalty to
do it this way verses starting a Flink instance in local mode, or a full
fledged Flink cluster? For jobs with low workload, is there any down side
just to run it like a regular Java application?

A side question is, when running it with the mini cluster, I'd watch the
log messages. I find that the process seems to focus on one operator for a
while, then switch to another operator. For example, my simple Flink
application has a kafka source, a windowed aggregator, and an elasticsearch
sink. I'd see a lot of SourceFunction log messages pumping records into the
pipeline, then they (the logs) would stop for a while. I then see some
AggregateFunction logs as the records come in, and then SinkFunction logs
after that when the window is up. After that, this could be a pause of 5-10
seconds or longer, SourceFunction logs would show up again. Because of the
windowing operation, I expect the SinkFunction to fire once in a while but
I was expecting to see interleaving SourceFunction and AggregateFunction
logs showing all operators are being run at the same time, instead of logs
from the loop inside SourceFunction.run(), followed by logs from
AggregateFunction.add() method. Is this because I'm running it with the
mini cluster, or is this how things are expected to work?

Thanks in advance
Jack

Re: What if not to keep containers across attempts in HA setup?(Internet mail)

[Paul Lam]

Hi Devin,

Thanks for your reasoning! It’s consistent with my observation, and I fully 
agree with you.

Maybe we should create an issue for the Hadoop community if it is not fixed in 
the master branch.

Best,
Paul Lam


> 在 2018年11月15日，11:59，devinduan(段丁瑞)  写道：
> 
> Hi Paul:
> I have reviewed hadoop & Flink code.
> Flink setKeepContainersAcrossApplicationAttempts to true if you set flink 
> config high-availability to true. 
> 
> If you set yarn.resourcemanager.work-preserving-recovery.enabled false, 
> AM(JobManager) will be killed by ResouceManager and start anthoer AM when 
> failover.
> Flink setKeepContainersAcrossApplicationAttempts to true cause am start 
> from previous attempt. 
> 
> But current ResourceManager is new active, application is not set 
> AppAttempt. 
> So you will see NPE exception.
> I think hadoop comunnity should resolve this issue.
>> Best,
>> Devin
> 
> 
> 
> 
> 
> 发件人： Paul Lam 
> 发送时间： 2018-11-15 11:31
> 收件人： devinduan(段丁瑞) 
> 抄送： user 
> 主题： Re: What if not to keep containers across attempts in HA setup?(Internet 
> mail)
> Hi Devin,
> 
> Thanks for the pointer and it works!
> 
> But I have no permission to change the YARN conf in production environment by 
> myself and it would need an detailed 
> investigation of the Hadoop team to apply the new conf, so I’m still 
> interested in the difference between keeping and 
> not keeping containers across application attempts.
> 
> Best,
> Paul Lam
> 
> 
>> 在 2018年11月13日，17:27，devinduan(段丁瑞) > > 写道：
>> 
>> Hi Paul，
>> Could you check out your YARN property  
>> "yarn.resourcemanager.work-preserving-recovery.enabled"?
>> if value is false, set true and try it again.
>> Best,
>> Devin
>>  
>> 发件人： Paul Lam 
>> 发送时间： 2018-11-13 12:55
>> 收件人： Flink ML 
>> 主题： What if not to keep containers across attempts in HA setup?(Internet 
>> mail)
>> Hi,
>> 
>> Recently I found a bug on our YARN cluster that crashes the standby RM 
>> during a RM failover, and 
>> the bug is triggered by the keeping containers across attempts behavior of 
>> applications (see [1], a related 
>> issue but the patch is not exactly the fix, because the problem is not on 
>> recovery, but the attempt after 
>> the recovery).
>> 
>> Since YARN is a fundamental component and a maintenance of it would affect a 
>> lot users, as a last resort
>> I wonder if we could modify YarnClusterDescriptor and not to keep containers 
>> across attempts. 
>> 
>> IMHO, Flink application’s state is not dependent on YARN, so there is no 
>> state that must be recovered 
>> from the previous application attempt. In case of a application master 
>> failure, the taskmanagers can be 
>> shutdown and the cost is longer recovery time.
>> 
>> Please correct me if I’m wrong. Thank you!
>> 
>> [1]https://issues.apache.org/jira/browse/YARN-2823 
>> 
>> 
>> Best,
>> Paul Lam

Re:

[Steve Bistline]

More to the story the cluster kicked out this error:

Exception: java.lang.OutOfMemoryError thrown from the
UncaughtExceptionHandler in thread "flink-7"

Uncaught error from thread [Uncaught error from thread [Uncaught error from
thread [flink-scheduler-1]:
flink-akka.remote.default-remote-dispatcher-92]: flink-10]: Java heap
space, shutting down JVM since 'akka.jvm-exit-on-fatal-error' is enabled
for for ActorSystem[Java heap spaceflink, shutting down JVM since
'akka.jvm-exit-on-fatal-error' is enabled for for ActorSystem[Uncaught
error from thread [flink-akka.remote.default-remote-dispatcher-96]: Java
heap space, shutting down JVM since 'akka.jvm-exit-on-fatal-error' is
enabled for for ActorSystem[flink]

flink]

]

Java heap space, shutting down JVM since 'akka.jvm-exit-on-fatal-error' is
enabled for for ActorSystem[flink]

java.lang.OutOfMemoryError: Java heap space

java.lang.OutOfMemoryError: Java heap space

java.lang.OutOfMemoryError: Java heap space

at java.net.URLClassLoader.findClass(URLClassLoader.java:362)

at java.lang.ClassLoader.loadClass(ClassLoader.java:424)

at java.lang.ClassLoader.loadClass(ClassLoader.java:411)

at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)

at java.lang.ClassLoader.loadClass(ClassLoader.java:357)

at
akka.remote.transport.ProtocolStateActor.akka$remote$transport$ProtocolStateActor$$decodePdu(AkkaProtocolTransport.scala:658)

at
akka.remote.transport.ProtocolStateActor$$anonfun$4.applyOrElse(AkkaProtocolTransport.scala:474)

at
akka.remote.transport.ProtocolStateActor$$anonfun$4.applyOrElse(AkkaProtocolTransport.scala:469)

at
scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)

at akka.actor.FSM$class.processEvent(FSM.scala:663)

at
akka.remote.transport.ProtocolStateActor.processEvent(AkkaProtocolTransport.scala:285)

at akka.actor.FSM$class.akka$actor$FSM$$processMsg(FSM.scala:657)

at akka.actor.FSM$$anonfun$receive$1.applyOrElse(FSM.scala:651)

at akka.actor.Actor$class.aroundReceive(Actor.scala:502)

at
akka.remote.transport.ProtocolStateActor.aroundReceive(AkkaProtocolTransport.scala:285)

at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)

at akka.actor.ActorCell.invoke(ActorCell.scala:495)

at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)

at akka.dispatch.Mailbox.run(Mailbox.scala:224)

at akka.dispatch.Mailbox.exec(Mailbox.scala:234)

at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)

at
scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)

at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)

at
scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

java.lang.OutOfMemoryError: Java heap space



On Thu, Nov 15, 2018 at 8:23 PM Steve Bistline 
wrote:

>
> Well, hopefully the last problem with this project.
>
> Any thoughts would be appreciated
>
> =
>
> java.lang.RuntimeException: Exception occurred while processing valve output 
> watermark:
>   at 
> org.apache.flink.streaming.runtime.io.StreamInputProcessor$ForwardingValveOutputHandler.handleWatermark(StreamInputProcessor.java:265)
>   at 
> org.apache.flink.streaming.runtime.streamstatus.StatusWatermarkValve.findAndOutputNewMinWatermarkAcrossAlignedChannels(StatusWatermarkValve.java:189)
>   at 
> org.apache.flink.streaming.runtime.streamstatus.StatusWatermarkValve.inputWatermark(StatusWatermarkValve.java:111)
>   at 
> org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:184)
>   at 
> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:103)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:306)
>   at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703)
>   at java.lang.Thread.run(Thread.java:748)
> Caused by: java.lang.RuntimeException: Failure happened in filter function.
>   at org.apache.flink.cep.nfa.NFA.createDecisionGraph(NFA.java:704)
>   at org.apache.flink.cep.nfa.NFA.computeNextStates(NFA.java:500)
>   at org.apache.flink.cep.nfa.NFA.process(NFA.java:252)
>   at 
> org.apache.flink.cep.operator.AbstractKeyedCEPPatternOperator.processEvent(AbstractKeyedCEPPatternOperator.java:332)
>   at 
> org.apache.flink.cep.operator.AbstractKeyedCEPPatternOperator.lambda$onEventTime$0(AbstractKeyedCEPPatternOperator.java:235)
>   at 
> java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1382)
>   at 
> java.util.stream.ReferencePipeline$Head.forEachOrdered(ReferencePipeline.java:590)
>   at 
> org.apache.flink.cep.operator.AbstractKeyedCEPPatternOperator.onEventTime(AbstractKeyedCEPPatternOperator.java:234)
>   at 
> org.apache.flink.streaming.api.operators.HeapInternalTimerService.advanceWatermark(HeapInternalTimerService.java:288)
>   at 
> 

[no subject]

[Steve Bistline]

Well, hopefully the last problem with this project.

Any thoughts would be appreciated

=

java.lang.RuntimeException: Exception occurred while processing valve
output watermark:
at 
org.apache.flink.streaming.runtime.io.StreamInputProcessor$ForwardingValveOutputHandler.handleWatermark(StreamInputProcessor.java:265)
at 
org.apache.flink.streaming.runtime.streamstatus.StatusWatermarkValve.findAndOutputNewMinWatermarkAcrossAlignedChannels(StatusWatermarkValve.java:189)
at 
org.apache.flink.streaming.runtime.streamstatus.StatusWatermarkValve.inputWatermark(StatusWatermarkValve.java:111)
at 
org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:184)
at 
org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:103)
at 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:306)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.RuntimeException: Failure happened in filter function.
at org.apache.flink.cep.nfa.NFA.createDecisionGraph(NFA.java:704)
at org.apache.flink.cep.nfa.NFA.computeNextStates(NFA.java:500)
at org.apache.flink.cep.nfa.NFA.process(NFA.java:252)
at 
org.apache.flink.cep.operator.AbstractKeyedCEPPatternOperator.processEvent(AbstractKeyedCEPPatternOperator.java:332)
at 
org.apache.flink.cep.operator.AbstractKeyedCEPPatternOperator.lambda$onEventTime$0(AbstractKeyedCEPPatternOperator.java:235)
at 
java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1382)
at 
java.util.stream.ReferencePipeline$Head.forEachOrdered(ReferencePipeline.java:590)
at 
org.apache.flink.cep.operator.AbstractKeyedCEPPatternOperator.onEventTime(AbstractKeyedCEPPatternOperator.java:234)
at 
org.apache.flink.streaming.api.operators.HeapInternalTimerService.advanceWatermark(HeapInternalTimerService.java:288)
at 
org.apache.flink.streaming.api.operators.InternalTimeServiceManager.advanceWatermark(InternalTimeServiceManager.java:108)
at 
org.apache.flink.streaming.api.operators.AbstractStreamOperator.processWatermark(AbstractStreamOperator.java:734)
at 
org.apache.flink.streaming.runtime.io.StreamInputProcessor$ForwardingValveOutputHandler.handleWatermark(StreamInputProcessor.java:262)
... 7 more
Caused by: java.lang.NullPointerException

Re: Writing to S3

[Steve Bistline]

Hi Ken,

Thank you for the link... I had just found this and when I removed the
Hadoop dependencies ( not using in this project anyway ) things worked
fine.

Now just trying to figure out the credentials.

Thanks,

Steve

On Thu, Nov 15, 2018 at 7:12 PM Ken Krugler 
wrote:

> Hi Steve,
>
> This looks similar to
> https://stackoverflow.com/questions/52009823/flink-shaded-hadoop-s3-filesystems-still-requires-hdfs-default-and-hdfs-site-con
>
> I see that you have classes
> like org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.conf.Configuration
> in your jar.
>
> Which makes me think you’re not excluding those properly.
>
> — Ken
>
>
> On Nov 15, 2018, at 3:58 PM, Steve Bistline 
> wrote:
>
> I am trying to write out to S3 from Flink with the following code and
> getting the error below. Tried adding the parser as a dependency, etc.
>
> Any help would be appreciated
>
> Table result = tableEnv.sql("SELECT 'Alert ',t_sKey, t_deviceID, t_sValue 
> FROM SENSORS WHERE t_sKey='TEMPERATURE' AND t_sValue > " + 
> TEMPERATURE_THRESHOLD);
> tableEnv.toAppendStream(result, Row.class).print();
> // Write to S3 bucket
> DataStream dsRow = tableEnv.toAppendStream(result, Row.class);
> dsRow.writeAsText("s3://csv-lifeai-ai/flink-alerts");
>
>
> ===
>
> avax.xml.parsers.FactoryConfigurationError: Provider for class 
> javax.xml.parsers.DocumentBuilderFactory cannot be created
>   at 
> javax.xml.parsers.FactoryFinder.findServiceProvider(FactoryFinder.java:311)
>   at javax.xml.parsers.FactoryFinder.find(FactoryFinder.java:267)
>   at 
> javax.xml.parsers.DocumentBuilderFactory.newInstance(DocumentBuilderFactory.java:120)
>   at 
> org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.conf.Configuration.loadResource(Configuration.java:2565)
>   at 
> org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.conf.Configuration.loadResources(Configuration.java:2541)
>   at 
> org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.conf.Configuration.getProps(Configuration.java:2424)
>   at 
> org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.conf.Configuration.get(Configuration.java:1238)
>   at 
> org.apache.flink.fs.s3hadoop.S3FileSystemFactory.create(S3FileSystemFactory.java:98)
>   at 
> org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(FileSystem.java:397)
>   at org.apache.flink.core.fs.FileSystem.get(FileSystem.java:320)
>   at org.apache.flink.core.fs.Path.getFileSystem(Path.java:298)
>   at 
> org.apache.flink.api.common.io.FileOutputFormat.open(FileOutputFormat.java:218)
>   at 
> org.apache.flink.api.java.io.TextOutputFormat.open(TextOutputFormat.java:88)
>   at 
> org.apache.flink.streaming.api.functions.sink.OutputFormatSinkFunction.open(OutputFormatSinkFunction.java:64)
>   at 
> org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
>   at 
> org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
>   at 
> org.apache.flink.streaming.api.operators.StreamSink.open(StreamSink.java:48)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:420)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:296)
>   at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703)
>   at java.lang.Thread.run(Thread.java:748)
>
>
> --
> Ken Krugler
> +1 530-210-6378
> http://www.scaleunlimited.com
> Custom big data solutions & training
> Flink, Solr, Hadoop, Cascading & Cassandra
>
>

Re: Writing to S3

[Ken Krugler]

Hi Steve,

This looks similar to 
https://stackoverflow.com/questions/52009823/flink-shaded-hadoop-s3-filesystems-still-requires-hdfs-default-and-hdfs-site-con
 


I see that you have classes like 
org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.conf.Configuration in 
your jar.

Which makes me think you’re not excluding those properly.

— Ken


> On Nov 15, 2018, at 3:58 PM, Steve Bistline  wrote:
> 
> I am trying to write out to S3 from Flink with the following code and getting 
> the error below. Tried adding the parser as a dependency, etc.
> 
> Any help would be appreciated
> 
> Table result = tableEnv.sql("SELECT 'Alert ',t_sKey, t_deviceID, t_sValue 
> FROM SENSORS WHERE t_sKey='TEMPERATURE' AND t_sValue > " + 
> TEMPERATURE_THRESHOLD);
> tableEnv.toAppendStream(result, Row.class).print();
> // Write to S3 bucket
> DataStream dsRow = tableEnv.toAppendStream(result, Row.class);
> dsRow.writeAsText("s3://csv-lifeai-ai/flink-alerts");
> 
> ===
> 
> avax.xml.parsers.FactoryConfigurationError: Provider for class 
> javax.xml.parsers.DocumentBuilderFactory cannot be created
>   at 
> javax.xml.parsers.FactoryFinder.findServiceProvider(FactoryFinder.java:311)
>   at javax.xml.parsers.FactoryFinder.find(FactoryFinder.java:267)
>   at 
> javax.xml.parsers.DocumentBuilderFactory.newInstance(DocumentBuilderFactory.java:120)
>   at 
> org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.conf.Configuration.loadResource(Configuration.java:2565)
>   at 
> org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.conf.Configuration.loadResources(Configuration.java:2541)
>   at 
> org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.conf.Configuration.getProps(Configuration.java:2424)
>   at 
> org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.conf.Configuration.get(Configuration.java:1238)
>   at 
> org.apache.flink.fs.s3hadoop.S3FileSystemFactory.create(S3FileSystemFactory.java:98)
>   at 
> org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(FileSystem.java:397)
>   at org.apache.flink.core.fs.FileSystem.get(FileSystem.java:320)
>   at org.apache.flink.core.fs.Path.getFileSystem(Path.java:298)
>   at 
> org.apache.flink.api.common.io.FileOutputFormat.open(FileOutputFormat.java:218)
>   at 
> org.apache.flink.api.java.io.TextOutputFormat.open(TextOutputFormat.java:88)
>   at 
> org.apache.flink.streaming.api.functions.sink.OutputFormatSinkFunction.open(OutputFormatSinkFunction.java:64)
>   at 
> org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
>   at 
> org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
>   at 
> org.apache.flink.streaming.api.operators.StreamSink.open(StreamSink.java:48)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:420)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:296)
>   at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703)
>   at java.lang.Thread.run(Thread.java:748)

--
Ken Krugler
+1 530-210-6378
http://www.scaleunlimited.com
Custom big data solutions & training
Flink, Solr, Hadoop, Cascading & Cassandra

Writing to S3

[Steve Bistline]

I am trying to write out to S3 from Flink with the following code and
getting the error below. Tried adding the parser as a dependency, etc.

Any help would be appreciated

Table result = tableEnv.sql("SELECT 'Alert ',t_sKey, t_deviceID,
t_sValue FROM SENSORS WHERE t_sKey='TEMPERATURE' AND t_sValue > " +
TEMPERATURE_THRESHOLD);
tableEnv.toAppendStream(result, Row.class).print();
// Write to S3 bucket
DataStream dsRow = tableEnv.toAppendStream(result, Row.class);
dsRow.writeAsText("s3://csv-lifeai-ai/flink-alerts");


===

avax.xml.parsers.FactoryConfigurationError: Provider for class
javax.xml.parsers.DocumentBuilderFactory cannot be created
at 
javax.xml.parsers.FactoryFinder.findServiceProvider(FactoryFinder.java:311)
at javax.xml.parsers.FactoryFinder.find(FactoryFinder.java:267)
at 
javax.xml.parsers.DocumentBuilderFactory.newInstance(DocumentBuilderFactory.java:120)
at 
org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.conf.Configuration.loadResource(Configuration.java:2565)
at 
org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.conf.Configuration.loadResources(Configuration.java:2541)
at 
org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.conf.Configuration.getProps(Configuration.java:2424)
at 
org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.conf.Configuration.get(Configuration.java:1238)
at 
org.apache.flink.fs.s3hadoop.S3FileSystemFactory.create(S3FileSystemFactory.java:98)
at 
org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(FileSystem.java:397)
at org.apache.flink.core.fs.FileSystem.get(FileSystem.java:320)
at org.apache.flink.core.fs.Path.getFileSystem(Path.java:298)
at 
org.apache.flink.api.common.io.FileOutputFormat.open(FileOutputFormat.java:218)
at 
org.apache.flink.api.java.io.TextOutputFormat.open(TextOutputFormat.java:88)
at 
org.apache.flink.streaming.api.functions.sink.OutputFormatSinkFunction.open(OutputFormatSinkFunction.java:64)
at 
org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
at 
org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
at 
org.apache.flink.streaming.api.operators.StreamSink.open(StreamSink.java:48)
at 
org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:420)
at 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:296)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703)
at java.lang.Thread.run(Thread.java:748)

Standalone HA cluster: Fatal error occurred in the cluster entrypoint.

[Olga Luganska]

Hello,

I am running flink 1.6.1 standalone HA cluster. Today I am unable to start 
cluster because of "Fatal error in cluster entrypoint"
(I used to see this error when running flink 1.5 version, after upgrade to 
1.6.1 (which had a fix for this bug) everything worked well for a while)

Question: what exactly needs to be done to clean "state handle store"?


2018-11-15 15:09:53,181 DEBUG 
org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor  - Fencing token 
not set: Ignoring message LocalFencedMessage(null, 
org.apache.flink.runtime.rpc.messages.RunAsync@21fd224c) because the fencing 
token is null.

2018-11-15 15:09:53,182 ERROR 
org.apache.flink.runtime.entrypoint.ClusterEntrypoint - Fatal error 
occurred in the cluster entrypoint.

java.lang.RuntimeException: org.apache.flink.util.FlinkException: Could not 
retrieve submitted JobGraph from state handle under 
/e13034f83a80072204facb2cec9ea6a3. This indicates that the retrieved state 
handle is broken. Try cleaning the state handle store.

at org.apache.flink.util.ExceptionUtils.rethrow(ExceptionUtils.java:199)

at 
org.apache.flink.util.function.FunctionUtils.lambda$uncheckedFunction$1(FunctionUtils.java:61)

at 
java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)

at 
java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)

at 
java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)

at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)

at 
akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)

at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)

at 
scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)

at 
scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)

at 
scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

Caused by: org.apache.flink.util.FlinkException: Could not retrieve submitted 
JobGraph from state handle under /e13034f83a80072204facb2cec9ea6a3. This 
indicates that the retrieved state handle is broken. Try cleaning the state 
handle store.

at 
org.apache.flink.runtime.jobmanager.ZooKeeperSubmittedJobGraphStore.recoverJobGraph(ZooKeeperSubmittedJobGraphStore.java:208)

at 
org.apache.flink.runtime.dispatcher.Dispatcher.recoverJob(Dispatcher.java:692)

at 
org.apache.flink.runtime.dispatcher.Dispatcher.recoverJobGraphs(Dispatcher.java:677)

at 
org.apache.flink.runtime.dispatcher.Dispatcher.recoverJobs(Dispatcher.java:658)

at 
org.apache.flink.runtime.dispatcher.Dispatcher.lambda$null$26(Dispatcher.java:817)

at 
org.apache.flink.util.function.FunctionUtils.lambda$uncheckedFunction$1(FunctionUtils.java:59)

... 9 more

Caused by: java.io.FileNotFoundException: 
/checkpoint_repo/ha/submittedJobGraphdd865937d674 (No such file or directory)

at java.io.FileInputStream.open0(Native Method)

at java.io.FileInputStream.open(FileInputStream.java:195)

at java.io.FileInputStream.(FileInputStream.java:138)

at 
org.apache.flink.core.fs.local.LocalDataInputStream.(LocalDataInputStream.java:50)

at 
org.apache.flink.core.fs.local.LocalFileSystem.open(LocalFileSystem.java:142)

at 
org.apache.flink.runtime.state.filesystem.FileStateHandle.openInputStream(FileStateHandle.java:68)

at 
org.apache.flink.runtime.state.RetrievableStreamStateHandle.openInputStream(RetrievableStreamStateHandle.java:64)

at 
org.apache.flink.runtime.state.RetrievableStreamStateHandle.retrieveState(RetrievableStreamStateHandle.java:57)

at 
org.apache.flink.runtime.jobmanager.ZooKeeperSubmittedJobGraphStore.recoverJobGraph(ZooKeeperSubmittedJobGraphStore.java:202)

... 14 more

2018-11-15 15:09:53,185 INFO  org.apache.flink.runtime.blob.TransientBlobCache  
- Shutting down BLOB cache


thank you,

Olga

Re: Join Dataset in stream

[Ken Krugler]

Hi Eric,

This sounds like a use case for BroadcastProcessFunction 

  You’d use the Cassandra dataset as the source for the broadcast stream, which 
is distributed to every parallel instance of your custom 
BroadcastProcessFunction. The input vectors are a partitioned stream that’s the 
other input to this function (via its processElement() method). The two streams 
get connected as a BroadcastConnectedStream 
.

Note that as of Flink 1.5 it’s also easy to maintain the broadcast state 
.

— Ken

> On Nov 14, 2018, at 11:32 PM, eric hoffmann  > wrote:
> 
> 
> Hi.
> I need to compute an euclidian distance between an input Vector and a full 
> dataset stored in Cassandra and keep the n lowest value. The Cassandra 
> dataset is evolving (mutable). I could do this on a batch job, but i will 
> have to triger it each time and the input are more like a slow stream, but 
> the computing need to be fast can i do this on a stream way? is there any 
> better solution ?
> Thx

--
Ken Krugler
+1 530-210-6378
http://www.scaleunlimited.com 
Custom big data solutions & training
Flink, Solr, Hadoop, Cascading & Cassandra

Re: Flink with parallelism 3 is running locally but not on cluster

[zavalit]

Hey, Dominik,
tnx for getting back.
i've posted also by stackoverflow and David Anderson gave a good tipp where
to look.
https://stackoverflow.com/questions/53282967/run-flink-with-parallelism-more-than-1/53289840
issues is resolved, everything is running.

thx. again



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Enabling Flink’s checkpointing

[Olga Luganska]

Hello,

By reading Flink documentation I see that to enable checkpointing we need to:

1. Enable checkpointing at the execution environment.

2. Make sure that your source/sink implements either CheckpointedFunction or 
ListCheckpointed interfaces?

Is #2 a must, and how checkpointing mechanism is affected if your source does 
not implement mentioned above interfaces?

(I see example of using RuntimeContext accessing keyed state)

Please explain.
Thank you very much,
Olga


Sent from my iPhone

Re: Run Time Exception

[marzieh ghasemi]

Thank you!

On Thu, Nov 15, 2018 at 4:18 PM Chesnay Schepler  wrote:

> I'm highly encouraging you to read through the examples and Batch API
> documentation:
>
>
> https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/batch/#example-program
>
> https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/batch/dataset_transformations.html#dataset-transformations
>
> TL;DR: apply a map function to the transactions DataSet
>
> On 15.11.2018 13:42, Mar_zieh wrote:
> > Hello guys
> >
> > I have wrote this code. I could print the whole file, but I want to read
> the
> > file line by line and process each line separately. would you please help
> > how I can do that?
> >
> >   ExecutionEnvironment env =
> ExecutionEnvironment.createLocalEnvironment();
> >  DataSet transactions =
> > env.readTextFile("/home/cfms11/Downloads/Data.csv");
> >  transactions.print();
> >
> > Thank you in advance.
> >
> >
> >
> > --
> > Sent from:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
> >
>
>

Re: Run Time Exception

[Chesnay Schepler]

I'm highly encouraging you to read through the examples and Batch API 
documentation:


https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/batch/#example-program
https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/batch/dataset_transformations.html#dataset-transformations

TL;DR: apply a map function to the transactions DataSet

On 15.11.2018 13:42, Mar_zieh wrote:

Hello guys

I have wrote this code. I could print the whole file, but I want to read the
file line by line and process each line separately. would you please help
how I can do that?

  ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment();
 DataSet transactions =
env.readTextFile("/home/cfms11/Downloads/Data.csv");
 transactions.print();

Thank you in advance.



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Run Time Exception

[Mar_zieh]

Hello guys

I have wrote this code. I could print the whole file, but I want to read the
file line by line and process each line separately. would you please help
how I can do that?

 ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment();
DataSet transactions =
env.readTextFile("/home/cfms11/Downloads/Data.csv");
transactions.print();

Thank you in advance. 



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Run Time Exception

[Mar_zieh]

Hello guys

I have wrote this code. I could print the whole file, but I want to read the
file line by line and process each line separately. would you please help
how I can do that?

 ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment();
DataSet transactions =
env.readTextFile("/home/cfms11/Downloads/Data.csv");
transactions.print();





--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Partitioning by composite key, But type and number of keys are dynamic

[Chesnay Schepler]

Why don't you calculate the hashCode for each field, combine them and 
use that as the key?
You won't get around calculating /something /for each field and 
combining the result.


On 15.11.2018 11:02, Gaurav Luthra wrote:

Hi Chesnay,

My End user will be aware about the fields of "input records"
(GenericRecord). In configuration my end user only will tell me the name and
number of the fields, based on these fields of GenericRecord I will have to
partition the DataStream and make Keyed Stream.

Currently, I have implemented my solution using KeySelector() function. And
I have converted all the fields mentioned by user into strings, concatenated
these strings (for all the fields for which user want keyBy() means user
wants partitioning) and returned single string from KeySelector() function.
And partitioning will be happening based on this concatenated string.
See the example below.

 dataStream.keyBy(record ->
 {
 return
record.get("area").toString()+record.get("age").toString();
 });


But, I am looking for better solution. As I do not want to convert every
field to string.




--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Partitioning by composite key, But type and number of keys are dynamic

[Gaurav Luthra]

Hi Chesnay,

My End user will be aware about the fields of "input records"
(GenericRecord). In configuration my end user only will tell me the name and
number of the fields, based on these fields of GenericRecord I will have to
partition the DataStream and make Keyed Stream.

Currently, I have implemented my solution using KeySelector() function. And
I have converted all the fields mentioned by user into strings, concatenated
these strings (for all the fields for which user want keyBy() means user
wants partitioning) and returned single string from KeySelector() function. 
And partitioning will be happening based on this concatenated string.
See the example below.

dataStream.keyBy(record ->
{
return
record.get("area").toString()+record.get("age").toString();
});


But, I am looking for better solution. As I do not want to convert every
field to string.




--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Field could not be resolved by the field mapping when using kafka connector

[Dominik Wosiński]

Hey,
Thanks for the info, I haven't noticed that.
I was just going through older messages with no responses.

Best Regards,
Dom.

Re: Flink with parallelism 3 is running locally but not on cluster

[Dominik Wosiński]

PS.
Could You also post the whole log for the application run ??

Best Regards,
Dom.

czw., 15 lis 2018 o 11:04 Dominik Wosiński  napisał(a):

> Hey,
>
> DId You try to run any other job on your setup? Also, could You please
> tell what are the sources you are trying to use, do all messages come from
> Kafka??
> From the first look, it seems that the JobManager can't connect to one of
> the TaskManagers.
>
>
> Best Regards,
> Dom.
>
> pon., 12 lis 2018 o 17:12 zavalit  napisał(a):
>
>> Hi,
>> may be i just missing smth, but i just have no more ideas where to look.
>>
>> here is an screen of the failed state
>> <
>> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/file/t1383/Bildschirmfoto_2018-11-12_um_16.png>
>>
>>
>> i read messages from 2 sources, make a join based on a common key and sink
>> it all in a kafka.
>>
>>   val env = StreamExecutionEnvironment.getExecutionEnvironment
>>   env.setParallelism(3)
>>   ...
>>   source1
>>  .keyBy(_.searchId)
>>  .connect(source2.keyBy(_.searchId))
>>  .process(new SearchResultsJoinFunction)
>>  .addSink(KafkaSink.sink)
>>
>> so it perfectly works when launch it locally. when i deploy it to 1 job
>> manager and 3 taskmanagers and get every Task in "RUNNING" state, after 2
>> minutes (when nothing is comming to sink) one of the taskmanagers gets
>> following in log:
>>
>>  Flat Map (1/3) (9598c11996f4b52a2e2f9f532f91ff66) switched from RUNNING
>> to
>> FAILED.
>> java.io.IOException: Connecting the channel failed: Connecting to remote
>> task manager + 'flink-taskmanager-11-dn9cj/10.81.27.84:37708' has failed.
>> This might indicate that the remote task manager has been lost.
>> at
>> org.apache.flink.runtime.io
>> .network.netty.PartitionRequestClientFactory$ConnectingChannel.waitForChannel(PartitionRequestClientFactory.java:196)
>> at
>> org.apache.flink.runtime.io
>> .network.netty.PartitionRequestClientFactory$ConnectingChannel.access$000(PartitionRequestClientFactory.java:133)
>> at
>> org.apache.flink.runtime.io
>> .network.netty.PartitionRequestClientFactory.createPartitionRequestClient(PartitionRequestClientFactory.java:85)
>> at
>> org.apache.flink.runtime.io
>> .network.netty.NettyConnectionManager.createPartitionRequestClient(NettyConnectionManager.java:60)
>> at
>> org.apache.flink.runtime.io
>> .network.partition.consumer.RemoteInputChannel.requestSubpartition(RemoteInputChannel.java:166)
>> at
>> org.apache.flink.runtime.io
>> .network.partition.consumer.SingleInputGate.requestPartitions(SingleInputGate.java:494)
>> at
>> org.apache.flink.runtime.io
>> .network.partition.consumer.SingleInputGate.getNextBufferOrEvent(SingleInputGate.java:525)
>> at
>> org.apache.flink.runtime.io
>> .network.partition.consumer.SingleInputGate.getNextBufferOrEvent(SingleInputGate.java:508)
>> at
>>
>> org.apache.flink.streaming.runtime.io.BarrierTracker.getNextNonBlocked(BarrierTracker.java:94)
>> at
>> org.apache.flink.streaming.runtime.io
>> .StreamInputProcessor.processInput(StreamInputProcessor.java:209)
>> at
>>
>> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:105)
>> at
>>
>> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)
>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
>> at java.lang.Thread.run(Thread.java:748)
>> Caused by:
>> org.apache.flink.runtime.io
>> .network.netty.exception.RemoteTransportException:
>> Connecting to remote task manager +
>> 'flink-taskmanager-11-dn9cj/10.81.27.84:37708' has failed. This might
>> indicate that the remote task manager has been lost.
>> at
>> org.apache.flink.runtime.io
>> .network.netty.PartitionRequestClientFactory$ConnectingChannel.operationComplete(PartitionRequestClientFactory.java:219)
>> at
>> org.apache.flink.runtime.io
>> .network.netty.PartitionRequestClientFactory$ConnectingChannel.operationComplete(PartitionRequestClientFactory.java:133)
>> at
>>
>> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:511)
>> at
>>
>> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:504)
>> at
>>
>> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:483)
>> at
>>
>> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:424)
>> at
>>
>> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:121)
>> at
>>
>> org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe$1.run(AbstractNioChannel.java:269)
>> at
>>
>> org.apache.flink.shaded.netty4.io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38)
>> 

Re: Flink with parallelism 3 is running locally but not on cluster

[Dominik Wosiński]

Hey,

DId You try to run any other job on your setup? Also, could You please tell
what are the sources you are trying to use, do all messages come from
Kafka??
>From the first look, it seems that the JobManager can't connect to one of
the TaskManagers.


Best Regards,
Dom.

pon., 12 lis 2018 o 17:12 zavalit  napisał(a):

> Hi,
> may be i just missing smth, but i just have no more ideas where to look.
>
> here is an screen of the failed state
> <
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/file/t1383/Bildschirmfoto_2018-11-12_um_16.png>
>
>
> i read messages from 2 sources, make a join based on a common key and sink
> it all in a kafka.
>
>   val env = StreamExecutionEnvironment.getExecutionEnvironment
>   env.setParallelism(3)
>   ...
>   source1
>  .keyBy(_.searchId)
>  .connect(source2.keyBy(_.searchId))
>  .process(new SearchResultsJoinFunction)
>  .addSink(KafkaSink.sink)
>
> so it perfectly works when launch it locally. when i deploy it to 1 job
> manager and 3 taskmanagers and get every Task in "RUNNING" state, after 2
> minutes (when nothing is comming to sink) one of the taskmanagers gets
> following in log:
>
>  Flat Map (1/3) (9598c11996f4b52a2e2f9f532f91ff66) switched from RUNNING to
> FAILED.
> java.io.IOException: Connecting the channel failed: Connecting to remote
> task manager + 'flink-taskmanager-11-dn9cj/10.81.27.84:37708' has failed.
> This might indicate that the remote task manager has been lost.
> at
> org.apache.flink.runtime.io
> .network.netty.PartitionRequestClientFactory$ConnectingChannel.waitForChannel(PartitionRequestClientFactory.java:196)
> at
> org.apache.flink.runtime.io
> .network.netty.PartitionRequestClientFactory$ConnectingChannel.access$000(PartitionRequestClientFactory.java:133)
> at
> org.apache.flink.runtime.io
> .network.netty.PartitionRequestClientFactory.createPartitionRequestClient(PartitionRequestClientFactory.java:85)
> at
> org.apache.flink.runtime.io
> .network.netty.NettyConnectionManager.createPartitionRequestClient(NettyConnectionManager.java:60)
> at
> org.apache.flink.runtime.io
> .network.partition.consumer.RemoteInputChannel.requestSubpartition(RemoteInputChannel.java:166)
> at
> org.apache.flink.runtime.io
> .network.partition.consumer.SingleInputGate.requestPartitions(SingleInputGate.java:494)
> at
> org.apache.flink.runtime.io
> .network.partition.consumer.SingleInputGate.getNextBufferOrEvent(SingleInputGate.java:525)
> at
> org.apache.flink.runtime.io
> .network.partition.consumer.SingleInputGate.getNextBufferOrEvent(SingleInputGate.java:508)
> at
>
> org.apache.flink.streaming.runtime.io.BarrierTracker.getNextNonBlocked(BarrierTracker.java:94)
> at
> org.apache.flink.streaming.runtime.io
> .StreamInputProcessor.processInput(StreamInputProcessor.java:209)
> at
>
> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:105)
> at
>
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
> at java.lang.Thread.run(Thread.java:748)
> Caused by:
> org.apache.flink.runtime.io
> .network.netty.exception.RemoteTransportException:
> Connecting to remote task manager +
> 'flink-taskmanager-11-dn9cj/10.81.27.84:37708' has failed. This might
> indicate that the remote task manager has been lost.
> at
> org.apache.flink.runtime.io
> .network.netty.PartitionRequestClientFactory$ConnectingChannel.operationComplete(PartitionRequestClientFactory.java:219)
> at
> org.apache.flink.runtime.io
> .network.netty.PartitionRequestClientFactory$ConnectingChannel.operationComplete(PartitionRequestClientFactory.java:133)
> at
>
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:511)
> at
>
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:504)
> at
>
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:483)
> at
>
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:424)
> at
>
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:121)
> at
>
> org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe$1.run(AbstractNioChannel.java:269)
> at
>
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38)
> at
>
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:125)
> at
>
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
> 

Re: Field could not be resolved by the field mapping when using kafka connector

[Chesnay Schepler]

This issue was already resolved in another thread by the same author.

On 15.11.2018 10:52, Dominik Wosiński wrote:

Hey,

Could You please show a sample data that You want to process? This 
would help in verifying the issue.


Best Regards,
Dom.

wt., 13 lis 2018 o 13:58 Jeff Zhang > napisał(a):


Hi,

I hit the following error when I try to use kafka connector in
flink table api. There's very little document about how to use
kafka connector in flink table api, could anyone help me on that ?
Thanks

Exception in thread "main"
org.apache.flink.table.api.ValidationException: Field 'event_ts'
could not be resolved by the field mapping.
at

org.apache.flink.table.sources.TableSourceUtil$.org$apache$flink$table$sources$TableSourceUtil$$resolveInputField(TableSourceUtil.scala:491)
at

org.apache.flink.table.sources.TableSourceUtil$$anonfun$org$apache$flink$table$sources$TableSourceUtil$$resolveInputFields$1.apply(TableSourceUtil.scala:521)
at

org.apache.flink.table.sources.TableSourceUtil$$anonfun$org$apache$flink$table$sources$TableSourceUtil$$resolveInputFields$1.apply(TableSourceUtil.scala:521)
at

scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at

scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at

scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
at
scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:186)
at

org.apache.flink.table.sources.TableSourceUtil$.org$apache$flink$table$sources$TableSourceUtil$$resolveInputFields(TableSourceUtil.scala:521)
at

org.apache.flink.table.sources.TableSourceUtil$.validateTableSource(TableSourceUtil.scala:127)
at

org.apache.flink.table.plan.schema.StreamTableSourceTable.(StreamTableSourceTable.scala:33)
at

org.apache.flink.table.api.StreamTableEnvironment.registerTableSourceInternal(StreamTableEnvironment.scala:150)
at

org.apache.flink.table.api.TableEnvironment.registerTableSource(TableEnvironment.scala:541)
at

org.apache.flink.table.descriptors.ConnectTableDescriptor.registerTableSource(ConnectTableDescriptor.scala:47)
at

org.apache.flink.table.descriptors.ConnectTableDescriptor.registerTableSourceAndSink(ConnectTableDescriptor.scala:68)

And here's the source code:



  case class Record(status:String, direction:String,var event_ts: Timestamp)


   def main(args: Array[String]): Unit = {
 val senv = StreamExecutionEnvironment.getExecutionEnvironment 
senv.setParallelism(1)
 senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

 val data: DataStream[Record] = ...
 val tEnv = TableEnvironment.getTableEnvironment(senv)tEnv
   // declare the external system to connect to .connect(
   new Kafka()
 .version("0.11")
 .topic("processed5.events")
 .startFromEarliest()
 .property("zookeeper.connect","localhost:2181")
 .property("bootstrap.servers","localhost:9092"))
   .withFormat(new Json()
 .failOnMissingField(false)
 .deriveSchema()
   )
   .withSchema(
 new Schema()
   .field("status", Types.STRING)
   .field("direction", Types.STRING)
   .field("event_ts", Types.SQL_TIMESTAMP).rowtime(
   new 
Rowtime().timestampsFromField("event_ts").watermarksPeriodicAscending())
   )

   // specify the update-mode for streaming tables .inAppendMode()

   // register as source, sink, or both and under a name 
.registerTableSourceAndSink("MyUserTable");

 tEnv.fromDataStream(data).insertInto("MyUserTable")

Re: Field could not be resolved by the field mapping when using kafka connector

[Dominik Wosiński]

Hey,

Could You please show a sample data that You want to process? This would
help in verifying the issue.

Best Regards,
Dom.

wt., 13 lis 2018 o 13:58 Jeff Zhang  napisał(a):

> Hi,
>
> I hit the following error when I try to use kafka connector in flink table
> api. There's very little document about how to use kafka connector in flink
> table api, could anyone help me on that ? Thanks
>
> Exception in thread "main" org.apache.flink.table.api.ValidationException:
> Field 'event_ts' could not be resolved by the field mapping.
> at
> org.apache.flink.table.sources.TableSourceUtil$.org$apache$flink$table$sources$TableSourceUtil$$resolveInputField(TableSourceUtil.scala:491)
> at
> org.apache.flink.table.sources.TableSourceUtil$$anonfun$org$apache$flink$table$sources$TableSourceUtil$$resolveInputFields$1.apply(TableSourceUtil.scala:521)
> at
> org.apache.flink.table.sources.TableSourceUtil$$anonfun$org$apache$flink$table$sources$TableSourceUtil$$resolveInputFields$1.apply(TableSourceUtil.scala:521)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at
> scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
> at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
> at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
> at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:186)
> at
> org.apache.flink.table.sources.TableSourceUtil$.org$apache$flink$table$sources$TableSourceUtil$$resolveInputFields(TableSourceUtil.scala:521)
> at
> org.apache.flink.table.sources.TableSourceUtil$.validateTableSource(TableSourceUtil.scala:127)
> at
> org.apache.flink.table.plan.schema.StreamTableSourceTable.(StreamTableSourceTable.scala:33)
> at
> org.apache.flink.table.api.StreamTableEnvironment.registerTableSourceInternal(StreamTableEnvironment.scala:150)
> at
> org.apache.flink.table.api.TableEnvironment.registerTableSource(TableEnvironment.scala:541)
> at
> org.apache.flink.table.descriptors.ConnectTableDescriptor.registerTableSource(ConnectTableDescriptor.scala:47)
> at
> org.apache.flink.table.descriptors.ConnectTableDescriptor.registerTableSourceAndSink(ConnectTableDescriptor.scala:68)
>
> And here's the source code:
>
>
>
>  case class Record(status: String, direction: String, var event_ts: Timestamp)
>
>
>   def main(args: Array[String]): Unit = {
> val senv = StreamExecutionEnvironment.getExecutionEnvironment
> senv.setParallelism(1)
> senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
>
> val data: DataStream[Record] = ...
> val tEnv = TableEnvironment.getTableEnvironment(senv)
> tEnv
>   // declare the external system to connect to
>   .connect(
>   new Kafka()
> .version("0.11")
> .topic("processed5.events")
> .startFromEarliest()
> .property("zookeeper.connect", "localhost:2181")
> .property("bootstrap.servers", "localhost:9092"))
>   .withFormat(new Json()
> .failOnMissingField(false)
> .deriveSchema()
>   )
>   .withSchema(
> new Schema()
>   .field("status", Types.STRING)
>   .field("direction", Types.STRING)
>   .field("event_ts", Types.SQL_TIMESTAMP).rowtime(
>   new 
> Rowtime().timestampsFromField("event_ts").watermarksPeriodicAscending())
>   )
>
>   // specify the update-mode for streaming tables
>   .inAppendMode()
>
>   // register as source, sink, or both and under a name
>   .registerTableSourceAndSink("MyUserTable");
>
> tEnv.fromDataStream(data).insertInto("MyUserTable")
>
>

Re: Rescaling Flink job from an External Checkpoint

[Chesnay Schepler]

The docs are worded that way since not all backends support it.

I believe rescaling does work for RocksDB checkpoints, but we do not 
provide any /guarantee /that this remains to be the case.


Basically, use at your own risk.

On 13.11.2018 13:24, suraj7 wrote:

Hi,

I'm using Flink 1.5 with RocksDB State Backend for Checkpointing. According
to the docs, Checkpoints do not support Job rescaling. However, when i tried
to restart a job from an externalised checkpoint with a different
parallelism, the job resumes correctly without any issues.

I would like to know if rescaling a Flink job from an external Checkpoint
based on RocksDB State Backend is reliable or are there any known
issues/drawbacks?

Thanks,
Suraj



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Could not find previous entry with key.

[Chesnay Schepler]

Can you provide us with the implementation of your Event and IoTEvent 
classes?


On 15.11.2018 06:10, Steve Bistline wrote:

Any thoughts on where to start with this error would be appreciated.

Caused by: java.lang.IllegalStateException: Could not find previous entry with key: first event, value: {"DEVICE_ID":f8a395a0-d3e2-11e8-b050-9779854d8172,"TIME_STAMP":11/15/2018 
02:29:30.343 
am,"TEMPERATURE":0.0,"HUMIDITY":"0.0","LIGHT_WHITE":0.0,"PROCX":0.0,"MOTION_DIRECTION":0,"MOTION_SPEED":0,"MOOD_STATE":0,"VISION_PEOPLE":0,"AUDIO1":0.0}
 and timestamp: 1542248971585. This can indicate that either you did not implement the equals() and hashCode() methods of your input elements properly or that the element belonging to that entry has 
been already pruned.
=
CODE HERE
=
//kinesisConsumerConfig.list(System.out); // Consume the data streams 
from AWS Kinesis stream DataStream dataStream = env.addSource(new FlinkKinesisConsumer<>(

pt.getRequired("stream"), new EventSchema(), 
kinesisConsumerConfig))
.name("Kinesis Stream Consumer"); System.out.printf("Print 
dataStream\n"); //dataStream.print(); DataStream kinesisStream = dataStream
.assignTimestampsAndWatermarks(new TimeLagWatermarkGenerator())
.map(event -> (IoTEvent) event); // Prints the mapped records from the Kinesis stream 
//kinesisStream.print(); //System.out.printf("Print kinesisStream\n"); Pattern pattern = Pattern

. begin("first event").subtype(IoTEvent.class)
.where(new IterativeCondition()
{
private static final long serialVersionUID = -6301755149429716724L; 
@Override
public boolean filter(IoTEvent value, Context ctx)throws 
Exception {
PatternConstants.MOTION_FIRST = value.getMotionDir(); return 
value.getMotionDir() != PatternConstants.MOTION_NA; }
})
.next("second")
.subtype(IoTEvent.class)
.where(new IterativeCondition() {
private static final long serialVersionUID 
=2392863109523984059L; @Override
public boolean filter(IoTEvent value, Context 
ctx)throws Exception {

return value.getMotionDir() != PatternConstants.MOTION_NA 
&& value.getMotionDir() != PatternConstants.MOTION_FIRST; }
})
.next("third")
.subtype(IoTEvent.class)
.where(new IterativeCondition() {
private static final long serialVersionUID 
=2392863109523984059L; @Override
public boolean filter(IoTEvent value, Context 
ctx)throws Exception {

return value.getMotionDir() != PatternConstants.MOTION_NA 
&& value.getMotionDir() == PatternConstants.MOTION_FIRST; }
})
.next("fourth")
.subtype(IoTEvent.class)
.where(new IterativeCondition() {
private static final long serialVersionUID 
=2392863109523984059L; @Override
public boolean filter(IoTEvent value, Context 
ctx)throws Exception {

return value.getMotionDir() != PatternConstants.MOTION_NA 
&& value.getMotionDir() != PatternConstants.MOTION_FIRST; }
})
.within(Time.seconds(10));

Re: Get savepoint status fails - Flink 1.6.2

[miki haiat]

Can you share some logs

On Thu, Nov 15, 2018 at 10:46 AM PedroMrChaves 
wrote:

> Hello,
>
> I've tried with different (jobId, triggerId) pairs but it doesn't work.
>
>
> Regards,
> Pedro Chaves.
>
>
>
> -
> Best Regards,
> Pedro Chaves
> --
> Sent from:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
>

Re: Get savepoint status fails - Flink 1.6.2

[PedroMrChaves]

Hello,

I've tried with different (jobId, triggerId) pairs but it doesn't work.


Regards,
Pedro Chaves.



-
Best Regards,
Pedro Chaves
--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/