Re: keyBy and parallelism

[Hao Sun]

>From what I learnt, you have to control parallelism your self. You can set
parallelism on operator or set default one through flink-config.yaml.
I might be wrong.

On Wed, Apr 11, 2018 at 2:16 PM Christophe Jolif  wrote:

> Hi all,
>
> Imagine I have a default parallelism of 16 and I do something like
>
> stream.keyBy("something").flatMap()
>
> Now let's imagine I have less than 16 keys, maybe 8.
>
> How many parallel executions of the flatMap function will I get? 8 because
> I have 8 keys, or 16 because I have default parallelism at 16?
>
> (and I will have follow up questions depending on the answer I suspect ;))
>
> Thanks,
> --
> Christophe
>

keyBy and parallelism

[Christophe Jolif]

Hi all,

Imagine I have a default parallelism of 16 and I do something like

stream.keyBy("something").flatMap()

Now let's imagine I have less than 16 keys, maybe 8.

How many parallel executions of the flatMap function will I get? 8 because
I have 8 keys, or 16 because I have default parallelism at 16?

(and I will have follow up questions depending on the answer I suspect ;))

Thanks,
-- 
Christophe

State management and heap usage

[TechnoMage]

I am pretty new to flink and have an initial streaming job working both locally 
and remotely.  But, both ways if the data volume is too high it runs out of 
heap.  I am using RichMapFunction to process multiple streams of data.  I 
assumed Flink would manage keeping state in ram when possible, and spill to 
RocksDB when it exceeded heap.

Is this correct?  If so are there configs I need to set to enable or tune this 
so it can run within a fixed memory size?

Michael

Re: Old JobManager lost its leadership in zk

[Steven Wu]

from Flink UI on jobmanager, sometimes I saw taskmanager connected and
heartbeat time got updated.


but then sometimes the taskmanager page become blank. maybe disconnected.



On Wed, Apr 11, 2018 at 1:31 PM, Steven Wu  wrote:

> Hi,
>
> After this error/exception, it seems that taskmanager never connects to
> jobmanager anymore.  Job stuck in failed state because there is not enough
> slots to recover the job.
>
> let's assume there was a temp glitch btw jobmanager and zk. would it cause
> such a permanent failure in Flink?
>
> I checked the zookeeper record.
> * leader zknode seems to have the correct info for "job_manager_lock"
> * I am not sure how to read the leaderlatch zknode
>
>
> A little more about the job
> * standalone cluster mode
> * 1 jobmanager
> * 1 taskmanager
>
> Thanks,
> Steven
>
> *2018-04-11 01:11:48,007 INFO  org.apache.flink.runtime.taskmanager.Task
> - Attempting to fail task externally Source:
> kafkasource -> Sink: s3sink (1/1)
> (5a7dba2e186b9fdaebb62bdd703dc7dc).2018-04-11 01:11:48,007 INFO
>  org.apache.flink.runtime.taskmanager.Task - Source:
> kafkasource -> Sink: s3sink (1/1) (5a7dba2e186b9fdaebb62bdd703dc7dc)
> switched from RUNNING to FAILED.java.lang.Exception: TaskManager
> akka://flink/user/taskmanager disconnects from JobManager
> akka.tcp://flink@1.2.3.4:42787/user/jobmanager
> : Old JobManager lost its
> leadership.at
> org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:1073)
>at org.apache.flink.runtime.taskmanager.TaskManager.org
> $apache$flink$runtime$taskmanager$TaskManager$$handleJobManagerLeaderAddress(TaskManager.scala:1467)
>at
> org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:277)
>at
> scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
>at
> org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:49)
>at
> scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
>at
> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
>at
> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
>at
> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)
>at
> org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
>at akka.actor.Actor$class.aroundReceive(Actor.scala:502)at
> org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:121)
>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)2018-04-11
> 01:11:48,011 INFO  org.apache.flink.runtime.taskmanager.Task
> - Triggering cancellation of task code Source:
> kafkasource -> Sink: s3sink (1/1)
> (5a7dba2e186b9fdaebb62bdd703dc7dc).2018-04-11 01:11:48,013 INFO
>  org.apache.flink.runtime.blob.PermanentBlobCache  - Shutting
> down BLOB cache*
>
>

Old JobManager lost its leadership in zk

[Steven Wu]

Hi,

After this error/exception, it seems that taskmanager never connects to
jobmanager anymore.  Job stuck in failed state because there is not enough
slots to recover the job.

let's assume there was a temp glitch btw jobmanager and zk. would it cause
such a permanent failure in Flink?

I checked the zookeeper record.
* leader zknode seems to have the correct info for "job_manager_lock"
* I am not sure how to read the leaderlatch zknode


A little more about the job
* standalone cluster mode
* 1 jobmanager
* 1 taskmanager

Thanks,
Steven

*2018-04-11 01:11:48,007 INFO  org.apache.flink.runtime.taskmanager.Task
- Attempting to fail task externally Source:
kafkasource -> Sink: s3sink (1/1)
(5a7dba2e186b9fdaebb62bdd703dc7dc).2018-04-11 01:11:48,007 INFO
 org.apache.flink.runtime.taskmanager.Task - Source:
kafkasource -> Sink: s3sink (1/1) (5a7dba2e186b9fdaebb62bdd703dc7dc)
switched from RUNNING to FAILED.java.lang.Exception: TaskManager
akka://flink/user/taskmanager disconnects from JobManager
akka.tcp://flink@1.2.3.4:42787/user/jobmanager
: Old JobManager lost its
leadership.at
org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:1073)
   at org.apache.flink.runtime.taskmanager.TaskManager.org
$apache$flink$runtime$taskmanager$TaskManager$$handleJobManagerLeaderAddress(TaskManager.scala:1467)
   at
org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:277)
   at
scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
   at
org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:49)
   at
scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
   at
org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
   at
org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
   at
scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)
   at
org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
   at akka.actor.Actor$class.aroundReceive(Actor.scala:502)at
org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:121)
   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)2018-04-11
01:11:48,011 INFO  org.apache.flink.runtime.taskmanager.Task
- Triggering cancellation of task code Source:
kafkasource -> Sink: s3sink (1/1)
(5a7dba2e186b9fdaebb62bdd703dc7dc).2018-04-11 01:11:48,013 INFO
 org.apache.flink.runtime.blob.PermanentBlobCache  - Shutting
down BLOB cache*

Re: Is Flink able to do real time stock market analysis?

[TechnoMage]

I am new to Flink so others may have more complete answer or correct me.

If you are counting the events in a tumbling window you will get output at the 
end of each tumbling window, so a running count of events/window.  It sounds 
like you want to compare the raw data to the smoothed data?  You can use a 
CoFlatMap to receive both streams and output any records you like, say a Tuple 
with the raw and smoothed value.  If you use a RichCoFlatMap you can track 
state, so you could keep a list of the last 20 or so raw and smoothed values so 
you can align them.

Michael

> On Apr 10, 2018, at 6:40 PM, Ivan Wang  wrote:
> 
> Hi all,
> 
> I've spent nearly 2 weeks trying to figure a solution to my requirement as 
> below. If anyone can advise, that would be great.
> 
> 1. There're going to be 2000 transactions per second as StreamRaw, I'm going 
> to tumbleWindow them as StreamA, let's say every 15 seconds. Then I'm going 
> to countWindow StreamA as StreamB, let's say every 20 events.
> 
> 2. For every event in  StreamRaw as E, I need to find exact one event in 
> StreamB which is earlier than E and closest to E. Then some comparison will 
> be proceeded. For example, if timestamp in E is 9:46:38, there should be an 
> event in StreamB with timestamp 9:46:30 because I use 15 seconds interval. 
> 
> I tried CEP using StreamRaw, however, I didn't figure out how to involve 
> StreamB and get the exact one event in condition method.
> 
> I tried tableAPI and SQL, it throws time attribute error during the second 
> window method. 
> 
> window(Tumble).group().select().window(Slide).group().select()
> 
> Seems there's no way to tell Flink the time attribute after the first 
> window.group(). I then tried to convert it into table first then leftoutJoin 
> them. But Flink tells me it's not supported.
> 
> Is Flink able to do this? If not, I'll go for other alternatives. Thanks 
> again if someone can help.
> 
> 
> 
> 
> 
> 
> 

Clarification on slots and efficiency

[Derek VerLee]

  
  


From the docs (
https://ci.apache.org/projects/flink/flink-docs-master/concepts/runtime.html
  )


  By adjusting the number of task slots, users can define how
subtasks are isolated from each other. Having one slot per
TaskManager means each task group runs in a separate JVM (which
can be started in a
separate container, for example). Having multiple slots
means more subtasks share the same JVM. Tasks in the same JVM
share TCP connections (via multiplexing) and
heartbeat messages. They may also share data sets and data
structures, thus reducing the per-task overhead.

Does this mean that if the same task and job is running in two
  slots on the same task-manager, that messages that happen to move
  between these slots will do so more efficiently, and avoid
  serialization overhead?



  

Json KAFKA producer

[Luigi Sgaglione]

Hi,

I'm trying to create a Flink example with kafka consumer and producer using
Json data format. In particular, I'm able to consume and process Json data
published on a Kafka topic, but not to publish the results.

The problem is that I don't know what is the serialization schema that
should be used to publish an ObjectNode (Jackson).


This is an excerpt of my test code

import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;


FlinkKafkaConsumer011 myConsumer = new
FlinkKafkaConsumer011<>("test2", new JSONDeserializationSchema(),
properties);
myConsumer.setStartFromEarliest();

DataStreamSource stream = env.addSource(myConsumer);
SingleOutputStreamOperator out1 = stream.filter(new
FilterFunction() {
private static final long serialVersionUID = 1L;

@Override
public boolean filter(ObjectNode arg0) throws Exception {
String temp=arg0.get("value").asText();
return (!temp.equals("1"));
}
});
FlinkKafkaProducer011 producer = new
FlinkKafkaProducer011("192.168.112.128:9092", "flinkOut", *XXX*
);
out1.addsink(producer);

Can you help me to understand how I can publish an ObjectNode?

Thanks

Re: How to add new Kafka topic to consumer

[Chesnay Schepler]

In 1.3 and below I believe this is not possible.

For 1.4 and above, please see 
https://ci.apache.org/projects/flink/flink-docs-master/dev/connectors/kafka.html#partition-discovery.


On 11.04.2018 14:33, chandresh pancholi wrote:

Hi,

Is there a way to add Kafka topic dynamically to stream?
For example there were Kafka topic named foo1, foo2, foo3 and task 
manager will start consuming events from all 3 topics. Now I create 
another two topic foo4 & bar1 in Kafka.


How will FlinkKafkaConsumer would read events from foo4 & bar1 topic?

Thanks

--
Chandresh Pancholi
Senior Software Engineer
Flipkart.com
Email-id:chandresh.panch...@flipkart.com 


Contact:08951803660

Re: How to add new Kafka topic to consumer

[Alexander Smirnov]

this feature has been implemented in 1.4.0, take a look at

https://issues.apache.org/jira/browse/FLINK-4022
https://ci.apache.org/projects/flink/flink-docs-release-1.4/dev/connectors/kafka.html#kafka-consumers-topic-and-partition-discovery


On Wed, Apr 11, 2018 at 3:33 PM chandresh pancholi <
chandreshpancholi...@gmail.com> wrote:

> Hi,
>
> Is there a way to add Kafka topic dynamically to stream?
> For example there were Kafka topic named foo1, foo2, foo3 and task manager
> will start consuming events from all 3 topics. Now I create another two
> topic foo4 & bar1 in Kafka.
>
> How will FlinkKafkaConsumer would read events from foo4 & bar1 topic?
>
> Thanks
>
> --
> Chandresh Pancholi
> Senior Software Engineer
> Flipkart.com
> Email-id:chandresh.panch...@flipkart.com
> Contact:08951803660
>

How to add new Kafka topic to consumer

[chandresh pancholi]

Hi,

Is there a way to add Kafka topic dynamically to stream?
For example there were Kafka topic named foo1, foo2, foo3 and task manager
will start consuming events from all 3 topics. Now I create another two
topic foo4 & bar1 in Kafka.

How will FlinkKafkaConsumer would read events from foo4 & bar1 topic?

Thanks

-- 
Chandresh Pancholi
Senior Software Engineer
Flipkart.com
Email-id:chandresh.panch...@flipkart.com
Contact:08951803660

Re: Allowed lateness + side output late data, what's included?

[Juho Autio]

Thanks!

On Wed, Apr 11, 2018 at 12:59 PM, Chesnay Schepler 
wrote:

> Data that arrives within the allowed lateness should not be written to the
> side output.
>
>
> On 11.04.2018 11:12, Juho Autio wrote:
>
> If I use a non-zero value for allowedLateness and also sideOutputLateData,
> does the late data output contain also the events that were triggered in
> the bounds of allowed lateness? By looking at the docs I can't be sure
> which way it is.
>
> Code example:
>
> .timeWindow(Time.days(1))
> .allowedLateness(Time.minutes(1))
> .sideOutputLateData(lateDataTag)
>
> Also I know that data that arrives more than 1 minute late is written to
> the late data side output.
>
> If some data arrives 0-1 minute late, I know that the time window triggers
> again with that data added. Is that data also written to the late data side
> output or not?
>
> Thanks!
>
>
>

Re: Allowed lateness + side output late data, what's included?

[Chesnay Schepler]

Data that arrives within the allowed lateness should not be written to 
the side output.


On 11.04.2018 11:12, Juho Autio wrote:
If I use a non-zero value for allowedLateness and also 
sideOutputLateData, does the late data output contain also the events 
that were triggered in the bounds of allowed lateness? By looking at 
the docs I can't be sure which way it is.


Code example:

.timeWindow(Time.days(1))
.allowedLateness(Time.minutes(1))
.sideOutputLateData(lateDataTag)

Also I know that data that arrives more than 1 minute late is written 
to the late data side output.


If some data arrives 0-1 minute late, I know that the time window 
triggers again with that data added. Is that data also written to the 
late data side output or not?


Thanks!

Allowed lateness + side output late data, what's included?

[Juho Autio]

If I use a non-zero value for allowedLateness and also sideOutputLateData,
does the late data output contain also the events that were triggered in
the bounds of allowed lateness? By looking at the docs I can't be sure
which way it is.

Code example:

.timeWindow(Time.days(1))
.allowedLateness(Time.minutes(1))
.sideOutputLateData(lateDataTag)

Also I know that data that arrives more than 1 minute late is written to
the late data side output.

If some data arrives 0-1 minute late, I know that the time window triggers
again with that data added. Is that data also written to the late data side
output or not?

Thanks!

Re: How to customize triggering of checkpoints?

[Chesnay Schepler]

Hello,

there is no way to manually trigger checkpoints or configure irregular 
intervals.


You will have to modify the CheckpointCoordinator 
 
and build Flink from source:


 * startCheckpointScheduler() should only schedule a one-time execution
   of the trigger
 * ScheduledTrigger#run() should reschedule itself
 o Something like:
 + triggerCheckpoint(System.currentTimeMillis(), true);
 + long msUntilNextCheckpoint = // insert logic here
 + timer.schedule(new ScheduledTrigger(),
   msUntilNextCheckpoint, TimeUnit.MILLISECONDS)

On 11.04.2018 05:15, syed wrote:

I am new to the flink environment and looking to analyze the triggering of
checkpoints. I am looking to trigger non-periodic checkpoints such that
checkpoint intervals are not of equal length, but not sure how can I do this
in Flink.

My specific query is;

(1) How can I trigger non-periodic checkpoints in Flink? I am looking to
trigger first checkpoint say after 10 seconds, the next checkpoint at say 25
seconds, third at 45 seconds and so on. Can I define my own function which
triggers non-periodic checkpoints and generates no-uniform checkpoint
intervals?
Thanks.




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

Re: Window with recent messages

[m@xi]

Hello there Krzystzof!

Thanks a lot for the answer. Sorry for the late reply. I can see the logic
behind custom window processing in Flink. 

Once, an incoming tuple arrives, you add a timer to it, which is going to
tick after "RatingExpiration" time units, as shown in your code. This, is
made *for each tuple*.

I have the following questions :

1 -- In my case, I do not have timestamps a-priori so I must append a
timestamp to the tuples as they arrive at the sources. Here [1] shows how to
assign timestamps to my data. Is this the correct way to do it? Also, what
type of watermarks is it better to assign? And what notion of time is it
more reasonable to use {EventTime, ProcessingTime, IngestionTime}?

2 -- The range of the sliding window equals to "RatingExpiration" if I am
correct. But, where is the slide of the sliding window defined? I guess the
slide has to do with the query, meaning each *s* time units evaluate the
query with the data residing in the range *r* last time units.

3 -- If I get to assign correctly the timestamps from above then it is
trivial, based also on your skeleton code, to simulate *time-based* sliding
windows. What about the case of *count-based* sliding windows???

Thanks a lot in advance. 

Best,
Max

[1] --
https://ci.apache.org/projects/flink/flink-docs-release-1.4/dev/event_timestamps_watermarks.html



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