Re: Flink SQL Stream Parser based on calcite

[Fabian Hueske]

The translation is done in multiple stages.

1. Parsing (syntax check)
2. Validation (semantic check)
3. Query optimization (rule and cost based)
4. Generation of physical plan, incl. code generation (DataStream program)

The final translation happens in the DataStream nodes, e.g., DataStreamCalc
[1].
I'd recommend to import the source code and to debug the translation
process.

I recently gave a talk about the high-level translation process [2].

Best,
Fabian

[1]
https://github.com/apache/flink/blob/master/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/datastream/DataStreamCalc.scala
[2]
http://www.slideshare.net/fhueske/taking-a-look-under-hood-of-apache-flinks-relational-apis





2016-10-17 18:42 GMT+02:00 PedroMrChaves :

> Thank you for the response.
>
> I'm not understanding where does something like this,
>
> /SELECT * WHERE action='denied' /
>
> gets translated to something similar in the Flink Stream API,
>
> /filter.(new FilterFunction() {
> public boolean filter(Event event) {
> return event.action.equals("denied");
> }
> });/
>
> or if that happens at all. My idea was to extend the library to support
> other unsupported
> calls like (TUMBLE -> timeWindow) but it's probably more complex than what
> I'm thinking.
>
> Regards.
>
>
>
>
> --
> View this message in context: http://apache-flink-user-
> mailing-list-archive.2336050.n4.nabble.com/Flink-SQL-
> Stream-Parser-based-on-calcite-tp9592p9596.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>

Re: FlinkML - Fail to execute QuickStart example

[Thomas FOURNIER]

Hi,

No problem I'm going to create a JIRA.

Regards
Thomas

2016-10-17 21:34 GMT+02:00 Theodore Vasiloudis <
theodoros.vasilou...@gmail.com>:

> That is my bad, I must have been testing against a private branch when
> writing the guide, the SVM as it stands only has a predict operation for
> Vector not LabeledVector.
>
> IMHO I would like to have a predict operator for LabeledVector for all
> predictors (that would just call the existing Vector prediction
> internally), but IIRC we decided to go with an Evaluate operator instead as
> written in the evaluation PR .
>
> I'll make a PR to fix the guide, any chance you can create a JIRA for this?
>
> Regards,
> Theodore
>
> On Mon, Oct 17, 2016 at 6:22 PM, Thomas FOURNIER <
> thomasfournier...@gmail.com> wrote:
>
>> Hi,
>>
>> Executing the following code (see QuickStart):
>>
>> val env = ExecutionEnvironment.getExecutionEnvironment
>> val survival = env.readCsvFile[(String, String, String, 
>> String)]("src/main/resources/haberman.data", ",")
>>
>>
>> val survivalLV = survival
>>   .map { tuple =>
>> val list = tuple.productIterator.toList
>> val numList = list.map(_.asInstanceOf[String].toDouble)
>> LabeledVector(numList(3), DenseVector(numList.take(3).toArray))
>>   }
>>
>>
>>
>> val astroTrain = MLUtils.readLibSVM(env, "src/main/resources/svmguide1")
>> val astroTest = MLUtils.readLibSVM(env, "src/main/resources/svmguide1.t")
>>
>>
>> val svm = SVM()
>>   .setBlocks(env.getParallelism)
>>   .setIterations(100)
>>   .setRegularization(0.001)
>>   .setStepsize(0.1)
>>   .setSeed(42)
>>
>> svm.fit(astroTrain)
>> svm.predict(astroTest)
>>
>>
>> I encounter the following error:
>>
>> Exception in thread "main" java.lang.RuntimeException: There is no 
>> PredictOperation defined for org.apache.flink.ml.classification.SVM which 
>> takes a DataSet[org.apache.flink.ml.common.LabeledVector] as input.
>>
>> Any idea ?
>>
>> Thanks
>>
>> Thomas
>>
>>
>>
>>
>>
>

Re: Task and Operator Monitoring via JMX / naming

[Philipp Bussche]

Thanks Chesnay.

I had a look at how the JMX representation looks like when I look at a Task
Manager which has one of the example Jobs deployed
(https://ci.apache.org/projects/flink/flink-docs-release-1.1/quickstart/run_example_quickstart.html)
and this looks correct.
I assume at this point that the naming gets confused because I am having
multiple sinks in my Job and more than one operator on the same stream.
Maybe this is not expected and I should only have one operator and one sink
per Job ? However the job itself does what it is supposed to so I would only
change this for the monitoring as it stands right now.
Also it seems to make a difference when things are happening in the job.
I had a print (sink) of the wikipedia source stream right at after the
source is read and after moving this print statement to the very end of the
job class the representation in JMX changes. I would expect the naming of
sinks and operators to be always the same regardless of when they happen, no
?

Thanks
Philipp



--
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Task-and-Operator-Monitoring-via-JMX-naming-tp9560p9600.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.

Re: FlinkML - Fail to execute QuickStart example

[Theodore Vasiloudis]

That is my bad, I must have been testing against a private branch when
writing the guide, the SVM as it stands only has a predict operation for
Vector not LabeledVector.

IMHO I would like to have a predict operator for LabeledVector for all
predictors (that would just call the existing Vector prediction
internally), but IIRC we decided to go with an Evaluate operator instead as
written in the evaluation PR .

I'll make a PR to fix the guide, any chance you can create a JIRA for this?

Regards,
Theodore

On Mon, Oct 17, 2016 at 6:22 PM, Thomas FOURNIER <
thomasfournier...@gmail.com> wrote:

> Hi,
>
> Executing the following code (see QuickStart):
>
> val env = ExecutionEnvironment.getExecutionEnvironment
> val survival = env.readCsvFile[(String, String, String, 
> String)]("src/main/resources/haberman.data", ",")
>
>
> val survivalLV = survival
>   .map { tuple =>
> val list = tuple.productIterator.toList
> val numList = list.map(_.asInstanceOf[String].toDouble)
> LabeledVector(numList(3), DenseVector(numList.take(3).toArray))
>   }
>
>
>
> val astroTrain = MLUtils.readLibSVM(env, "src/main/resources/svmguide1")
> val astroTest = MLUtils.readLibSVM(env, "src/main/resources/svmguide1.t")
>
>
> val svm = SVM()
>   .setBlocks(env.getParallelism)
>   .setIterations(100)
>   .setRegularization(0.001)
>   .setStepsize(0.1)
>   .setSeed(42)
>
> svm.fit(astroTrain)
> svm.predict(astroTest)
>
>
> I encounter the following error:
>
> Exception in thread "main" java.lang.RuntimeException: There is no 
> PredictOperation defined for org.apache.flink.ml.classification.SVM which 
> takes a DataSet[org.apache.flink.ml.common.LabeledVector] as input.
>
> Any idea ?
>
> Thanks
>
> Thomas
>
>
>
>
>

Re: Flink Metrics

[amir bahmanyari]

Hi colleagues,Is there a link that described Flink Matrices & provides example 
on how to utilize it pls?I really appreciate it...Cheers

  From: Till Rohrmann 
 To: user@flink.apache.org 
Cc: d...@flink.apache.org
 Sent: Monday, October 17, 2016 12:52 AM
 Subject: Re: Flink Metrics
   
Hi Govind,

I think the DropwizardMeterWrapper implementation is just a reference
implementation where it was decided to report the minute rate. You can
define your own meter class which allows to configure the rate interval
accordingly.

Concerning Timers, I think nobody requested this metric so far. If you
want, then you can open a JIRA issue and contribute it. The community would
really appreciate that.

Cheers,
Till
​

On Mon, Oct 17, 2016 at 5:26 AM, Govindarajan Srinivasaraghavan <
govindragh...@gmail.com> wrote:

> Hi,
>
> I am currently using flink 1.2 snapshot and instrumenting my pipeline with
> flink metrics. One small suggestion I have is currently the Meter interface
> only supports getRate() which is always the one minute rate.
>
> It would great if all the rates (1 min, 5 min & 15 min) are exposed to get
> a better picture in terms of performance.
>
> Also is there any reason why timers are not part of flink metrics core?
>
> Regards,
> Govind
>

   

Re: Testing a streaming data flow

[Stefan Richter]

Hi,

I think there are some things that could be helpful for testing your algorithm. 
From the top of my head, first thing is that you could try to test in a more 
„unit-testing“ style, i.e. just write small drivers that inject records to your 
UDFs and check if the output is as expected. 

Other than that, you should use event time instead of processing time for 
reproducible tests. With event time, there has to be no relationship from your 
source to some external timing (like System time) and no need to introduce 
sleeps to your code. You can find examples of this in several places in Flink’s 
tests. In particular, if you are interested in simulating sessions, you could 
take a look at 
org.apache.flink.test.windowing.sessionwindows.SessionWindowITCase and reuse 
some parts of it.

Best,
Stefan


> Am 17.10.2016 um 15:21 schrieb bart.kasterm...@kpn.com:
> 
> What are the standard approaches for testing a streaming algorithm?  I have 
> been
> able to come up with the below where I
> 
> 1) create a data source that emits events in bunches with set times so that I
> know the events will be in the same window,
> 2) end the stream with a mapWithState where the state checks if the expected
> elements pass by in the expected order.
> 
> This does not seem like the most robust way of doing this.  Suggestions?
> 
> Best,
> Bart
> 
> import java.io .{FileWriter, StringWriter}
> import java.util.{Calendar, Date, Properties}
> 
> import com.fasterxml.jackson.annotation.JsonIgnoreProperties
> import com.fasterxml.jackson.databind.ObjectMapper
> import com.fasterxml.jackson.module.scala.DefaultScalaModule
> import org.apache.flink.api.java.utils.ParameterTool
> import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
> import org.apache.flink.streaming.api.functions.source.SourceFunction
> import 
> org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
> import org.apache.flink.streaming.api.scala._
> import org.apache.flink.streaming.api.watermark.Watermark
> import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows
> import org.apache.flink.streaming.api.windowing.time.Time
> import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer09, 
> FlinkKafkaProducer09}
> import org.apache.flink.streaming.util.serialization.SimpleStringSchema
> import org.slf4j.{Logger, LoggerFactory}
> 
> import scala.math.max
> import scala.util.{Random, hashing}
> import java.time
> 
> 
> object SessionizePageviewsTT {
> 
>   val logger: Logger = LoggerFactory.getLogger("SessionizePageviewsTT") // 
> classOf doesn't work on an object
> 
>   def get_now_ms(): Long = {
> System.currentTimeMillis()
>   }
> 
>   def main(args: Array[String]) {
> 
> val env: StreamExecutionEnvironment = 
> StreamExecutionEnvironment.getExecutionEnvironment
> 
> val xs = 1 to 30
> def sourceFF(scx: SourceContext[Int]): Unit = {
>   var cur = 1
>   var now: Long = get_now_ms()
>   while (cur < 31) {
> // every 10 wait 10 seconds and then burst a bunch
> if (cur % 10 == 0) {
>   Thread.sleep(1)
>   now = get_now_ms()
> }
> println("emiting: " + cur + ", " + now)
> scx.collectWithTimestamp(cur, now)
> cur += 1
>   }
> }
> 
> val x: DataStream[Int] = env.addSource(sourceFF _)
> 
> val vals = List(45, 145);
> 
> def checkFF(xy: (Int, Int), s: Option[Int]): ((Int, Int), Option[Int]) = {
>   val idx = if (s.isDefined) s.get else 0
>   if (idx < vals.length) {
> if (xy._1 == vals(idx)) {
>   println("all ok")
> } else {
>   println("error error")
> }
>   }
> 
>   (xy, Some(idx + 1))
> }
> 
> x.map(x => (x,1)).keyBy(1).timeWindow(Time.seconds(10)).sum(0).keyBy(x => 
> 1).mapWithState(checkFF).print
> 
> env.execute(s"")
>   }
> }

Re: Flink SQL Stream Parser based on calcite

[PedroMrChaves]

Thank you for the response. 

I'm not understanding where does something like this,

/SELECT * WHERE action='denied' /

gets translated to something similar in the Flink Stream API,

/filter.(new FilterFunction() {
public boolean filter(Event event) {
return event.action.equals("denied");
}
});/

or if that happens at all. My idea was to extend the library to support
other unsupported
calls like (TUMBLE -> timeWindow) but it's probably more complex than what
I'm thinking. 

Regards.




--
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-SQL-Stream-Parser-based-on-calcite-tp9592p9596.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.

FlinkML - Fail to execute QuickStart example

[Thomas FOURNIER]

Hi,

Executing the following code (see QuickStart):

val env = ExecutionEnvironment.getExecutionEnvironment
val survival = env.readCsvFile[(String, String, String,
String)]("src/main/resources/haberman.data", ",")


val survivalLV = survival
  .map { tuple =>
val list = tuple.productIterator.toList
val numList = list.map(_.asInstanceOf[String].toDouble)
LabeledVector(numList(3), DenseVector(numList.take(3).toArray))
  }



val astroTrain = MLUtils.readLibSVM(env, "src/main/resources/svmguide1")
val astroTest = MLUtils.readLibSVM(env, "src/main/resources/svmguide1.t")


val svm = SVM()
  .setBlocks(env.getParallelism)
  .setIterations(100)
  .setRegularization(0.001)
  .setStepsize(0.1)
  .setSeed(42)

svm.fit(astroTrain)
svm.predict(astroTest)


I encounter the following error:

Exception in thread "main" java.lang.RuntimeException: There is no
PredictOperation defined for org.apache.flink.ml.classification.SVM
which takes a DataSet[org.apache.flink.ml.common.LabeledVector] as
input.

Any idea ?

Thanks

Thomas

Re: BoundedOutOfOrdernessTimestampExtractor and allowedlateness

[Yassine MARZOUGUI]

Hi Fabian,

Thank you very much for the great answer and example, I appreciate it!
It is all clear now.

Best,
Yassine

2016-10-17 16:29 GMT+02:00 Fabian Hueske :

> I have to extend my answer:
>
> The behavior allowedLateness that I described applies only if the window
> trigger calls FIRE when the window is evaluated (this is the default
> behavior of most triggers).
>
> In case the trigger calls FIRE_AND_PURGE, the state of the window is
> purged when the function is evaluated and late events are processed alone,
> i.e., in my example <12:09, G> would be processed without [A, B, C, D].
> When the allowed lateness is passed, all window state is purged regardless
> of the trigger.
>
> Best, Fabian
>
> 2016-10-17 16:24 GMT+02:00 Fabian Hueske :
>
>> Hi Yassine,
>>
>> the difference is the following:
>>
>> 1) The BoundedOutOfOrdernessTimestampExtractor is a built-in timestamp
>> extractor and watermark assigner.
>> A timestamp extractor tells Flink when an event happened, i.e., it
>> extracts a timestamp from the event. A watermark assigner tells Flink what
>> the current logical time is.
>> The BoundedOutOfOrdernessTimestampExtractor works as follows: When Flink
>> asks what the current time is, it returns the latest observed timestamp
>> minus the a configurable bound. This is the safety margin for late data.
>>  A record whose timestamp is lower than the last watermark is considered
>> to be late.
>>
>> 2) The allowedLateness parameter of time windows tells Flink how long to
>> keep state around after the window was evaluated.
>> If data arrives after the evaluation and before the allowedLateness has
>> passed, the window function is applied again and an update is sent out.
>>
>> Let's look at an example.
>> Assume you have a BOOTE with a 2 minute bound and a 10 minute tumbling
>> window that starts at 12:00 and ends at 12:10:
>>
>> If you have the following data:
>>
>> 12:01, A
>> 12:04, B
>> WM, 12:02 // 12:04 - 2 minutes
>> 12:02, C
>> 12:08, D
>> 12:14, E
>> WM, 12:12
>> 12:16, F
>> WM, 12:14 // 12:16 - 2 minutes
>> 12:09, G
>>
>> == no allowed lateness
>> The window operator forwards the logical time to 12:12 when it receives
>>  and evaluates the window which contains [A, B, C, D] at this
>> time and finally purges its state. <12:09, G> is later ignored.
>>
>> == allowed lateness of 3 minutes
>> The window operator evaluates the window when  is received,
>> but its state is not purged yet. The state is purged when  is
>> received (window fire time 12:10 + 3mins allowed lateness). <12:09, G> is
>> again ignored.
>>
>> == allowed lateness of 5 minutes
>> The window operator evaluates the window when  is received,
>> but its state is not purged yet. When <12:09, G> is received, the window is
>> again evaluated but this time with [A, B, C, D, G] and an update is sent
>> out. The state is purged when a watermark of >=12:15 is received.
>>
>> So, watermarks tell the Flink what time it is and allowed lateness tells
>> the system when state should be discarded and all later arriving data be
>> ignored.
>> These issue are related but not exactly the same thing. For instance you
>> can counter late data by increasing the bound or the lateness parameter.
>> Increasing the watermark bound will yield higher latencies as windows are
>> evaluated later.
>> Configuring allowedLateness will allow for earlier results, but you have
>> to cope with the updates downstream.
>>
>> Please let me know, if you have questions.
>>
>> Best, Fabian
>>
>>
>>
>>
>>
>>
>>
>>
>> 2016-10-17 11:52 GMT+02:00 Yassine MARZOUGUI :
>>
>>> Hi,
>>>
>>> I'm a bit confused about how Flink deals with late elements after the
>>> introduction of allowedlateness to windows. What is the difference between
>>> using a BoundedOutOfOrdernessTimestampExtractor(Time.seconds(X)) and
>>> allowedlateness(Time.seconds(X))? What if one is used and the other is
>>> not? and what if a different lateness is used in each one? Could you please
>>> clarify it on basis of a simple example? Thank you.
>>>
>>> Best,
>>> Yassine
>>>
>>
>>
>

Flink SQL Stream Parser based on calcite

[PedroMrChaves]

Hello,

I am pretty new to Apache Flink.

I am trying to figure out how does Flink parses an Apache Calcite sql query
to its own Streaming API in order to maybe extend it, because, as far as I
know, many operations are still being developed and not currently supported
(like TUMBLE windows). I need to be able to load rules from a file , like
so:

/tableEnv.sql([File])../

in order to do that I need a fully functional Streaming SQL parser. 

I am currently analyzing the StreamTableEnvironment class from github [1] in
order to understand the method sql but I can't figure out where does the
parsing happens.

Can someone point me in the right direction? 


[1] 
https://github.com/apache/flink/blob/d7b59d761601baba6765bb4fc407bcd9fd6a9387/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/StreamTableEnvironment.scala

  

Best Regards,
Pedro Chaves





--
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-SQL-Stream-Parser-based-on-calcite-tp9592.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.

Re: Flink SQL Stream Parser based on calcite

[Fabian Hueske]

Hi Pedro,

The sql() method calls the Calcite parser in line 129.

Best, Fabian

2016-10-17 16:43 GMT+02:00 PedroMrChaves :

> Hello,
>
> I am pretty new to Apache Flink.
>
> I am trying to figure out how does Flink parses an Apache Calcite sql query
> to its own Streaming API in order to maybe extend it, because, as far as I
> know, many operations are still being developed and not currently supported
> (like TUMBLE windows). I need to be able to load rules from a file , like
> so:
>
> /tableEnv.sql([File])../
>
> in order to do that I need a fully functional Streaming SQL parser.
>
> I am currently analyzing the StreamTableEnvironment class from github [1]
> in
> order to understand the method sql but I can't figure out where does the
> parsing happens.
>
> Can someone point me in the right direction?
>
>
> [1]
> https://github.com/apache/flink/blob/d7b59d761601baba6765bb4fc407bc
> d9fd6a9387/flink-libraries/flink-table/src/main/scala/
> org/apache/flink/api/table/StreamTableEnvironment.scala
>  d9fd6a9387/flink-libraries/flink-table/src/main/scala/
> org/apache/flink/api/table/StreamTableEnvironment.scala>
>
> Best Regards,
> Pedro Chaves
>
>
>
>
>
> --
> View this message in context: http://apache-flink-user-
> mailing-list-archive.2336050.n4.nabble.com/Flink-SQL-
> Stream-Parser-based-on-calcite-tp9592.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>

Re: BoundedOutOfOrdernessTimestampExtractor and allowedlateness

[Fabian Hueske]

I have to extend my answer:

The behavior allowedLateness that I described applies only if the window
trigger calls FIRE when the window is evaluated (this is the default
behavior of most triggers).

In case the trigger calls FIRE_AND_PURGE, the state of the window is purged
when the function is evaluated and late events are processed alone, i.e.,
in my example <12:09, G> would be processed without [A, B, C, D].
When the allowed lateness is passed, all window state is purged regardless
of the trigger.

Best, Fabian

2016-10-17 16:24 GMT+02:00 Fabian Hueske :

> Hi Yassine,
>
> the difference is the following:
>
> 1) The BoundedOutOfOrdernessTimestampExtractor is a built-in timestamp
> extractor and watermark assigner.
> A timestamp extractor tells Flink when an event happened, i.e., it
> extracts a timestamp from the event. A watermark assigner tells Flink what
> the current logical time is.
> The BoundedOutOfOrdernessTimestampExtractor works as follows: When Flink
> asks what the current time is, it returns the latest observed timestamp
> minus the a configurable bound. This is the safety margin for late data.
>  A record whose timestamp is lower than the last watermark is considered
> to be late.
>
> 2) The allowedLateness parameter of time windows tells Flink how long to
> keep state around after the window was evaluated.
> If data arrives after the evaluation and before the allowedLateness has
> passed, the window function is applied again and an update is sent out.
>
> Let's look at an example.
> Assume you have a BOOTE with a 2 minute bound and a 10 minute tumbling
> window that starts at 12:00 and ends at 12:10:
>
> If you have the following data:
>
> 12:01, A
> 12:04, B
> WM, 12:02 // 12:04 - 2 minutes
> 12:02, C
> 12:08, D
> 12:14, E
> WM, 12:12
> 12:16, F
> WM, 12:14 // 12:16 - 2 minutes
> 12:09, G
>
> == no allowed lateness
> The window operator forwards the logical time to 12:12 when it receives
>  and evaluates the window which contains [A, B, C, D] at this
> time and finally purges its state. <12:09, G> is later ignored.
>
> == allowed lateness of 3 minutes
> The window operator evaluates the window when  is received, but
> its state is not purged yet. The state is purged when  is
> received (window fire time 12:10 + 3mins allowed lateness). <12:09, G> is
> again ignored.
>
> == allowed lateness of 5 minutes
> The window operator evaluates the window when  is received, but
> its state is not purged yet. When <12:09, G> is received, the window is
> again evaluated but this time with [A, B, C, D, G] and an update is sent
> out. The state is purged when a watermark of >=12:15 is received.
>
> So, watermarks tell the Flink what time it is and allowed lateness tells
> the system when state should be discarded and all later arriving data be
> ignored.
> These issue are related but not exactly the same thing. For instance you
> can counter late data by increasing the bound or the lateness parameter.
> Increasing the watermark bound will yield higher latencies as windows are
> evaluated later.
> Configuring allowedLateness will allow for earlier results, but you have
> to cope with the updates downstream.
>
> Please let me know, if you have questions.
>
> Best, Fabian
>
>
>
>
>
>
>
>
> 2016-10-17 11:52 GMT+02:00 Yassine MARZOUGUI :
>
>> Hi,
>>
>> I'm a bit confused about how Flink deals with late elements after the
>> introduction of allowedlateness to windows. What is the difference between
>> using a BoundedOutOfOrdernessTimestampExtractor(Time.seconds(X)) and
>> allowedlateness(Time.seconds(X))? What if one is used and the other is
>> not? and what if a different lateness is used in each one? Could you please
>> clarify it on basis of a simple example? Thank you.
>>
>> Best,
>> Yassine
>>
>
>

Re: BoundedOutOfOrdernessTimestampExtractor and allowedlateness

[Fabian Hueske]

Hi Yassine,

the difference is the following:

1) The BoundedOutOfOrdernessTimestampExtractor is a built-in timestamp
extractor and watermark assigner.
A timestamp extractor tells Flink when an event happened, i.e., it extracts
a timestamp from the event. A watermark assigner tells Flink what the
current logical time is.
The BoundedOutOfOrdernessTimestampExtractor works as follows: When Flink
asks what the current time is, it returns the latest observed timestamp
minus the a configurable bound. This is the safety margin for late data.
 A record whose timestamp is lower than the last watermark is considered to
be late.

2) The allowedLateness parameter of time windows tells Flink how long to
keep state around after the window was evaluated.
If data arrives after the evaluation and before the allowedLateness has
passed, the window function is applied again and an update is sent out.

Let's look at an example.
Assume you have a BOOTE with a 2 minute bound and a 10 minute tumbling
window that starts at 12:00 and ends at 12:10:

If you have the following data:

12:01, A
12:04, B
WM, 12:02 // 12:04 - 2 minutes
12:02, C
12:08, D
12:14, E
WM, 12:12
12:16, F
WM, 12:14 // 12:16 - 2 minutes
12:09, G

== no allowed lateness
The window operator forwards the logical time to 12:12 when it receives
 and evaluates the window which contains [A, B, C, D] at this
time and finally purges its state. <12:09, G> is later ignored.

== allowed lateness of 3 minutes
The window operator evaluates the window when  is received, but
its state is not purged yet. The state is purged when  is
received (window fire time 12:10 + 3mins allowed lateness). <12:09, G> is
again ignored.

== allowed lateness of 5 minutes
The window operator evaluates the window when  is received, but
its state is not purged yet. When <12:09, G> is received, the window is
again evaluated but this time with [A, B, C, D, G] and an update is sent
out. The state is purged when a watermark of >=12:15 is received.

So, watermarks tell the Flink what time it is and allowed lateness tells
the system when state should be discarded and all later arriving data be
ignored.
These issue are related but not exactly the same thing. For instance you
can counter late data by increasing the bound or the lateness parameter.
Increasing the watermark bound will yield higher latencies as windows are
evaluated later.
Configuring allowedLateness will allow for earlier results, but you have to
cope with the updates downstream.

Please let me know, if you have questions.

Best, Fabian








2016-10-17 11:52 GMT+02:00 Yassine MARZOUGUI :

> Hi,
>
> I'm a bit confused about how Flink deals with late elements after the
> introduction of allowedlateness to windows. What is the difference between
> using a BoundedOutOfOrdernessTimestampExtractor(Time.seconds(X)) and
> allowedlateness(Time.seconds(X))? What if one is used and the other is
> not? and what if a different lateness is used in each one? Could you please
> clarify it on basis of a simple example? Thank you.
>
> Best,
> Yassine
>

Re: job failure with checkpointing enabled

[Aljoscha Krettek]

Ok, thanks for the update!

Let me know if you run into any more problems.

On Mon, 17 Oct 2016 at 14:40  wrote:

> HI Aljoscha,
>
>
>
> Thanks for the response.
>
>
>
> To answer your question, the base path did not exist.  But, I think I
> found the issue.  I believe I had some rogue task managers running.  As a
> troubleshooting step, I attempted to restart my cluster.  However, after
> shutting down the cluster I noticed that there were still task managers
> running on most of my nodes (and on the master).  Interestingly, on a
> second attempt to shut down the cluster, I received the message “No
> taskmanager daemon to stop on host…” for each of my nodes, even though I
> could see the flink processes running on these machines.   After manually
> killing these processes and restarting the cluster, the problem went away.
>
>
>
> So, my assumption is that on a previous attempt to bounce the cluster,
> these processes did not shut down cleanly.  Starting the cluster after that
> **may** have resulted in second instances of the task manager running on
> most nodes.  I’m not certain, however, and I haven’t yet been able to
> reproduce the issue.
>
>
>
>
>
>
>
>
>
>
>
> *From: *Aljoscha Krettek 
> *Reply-To: *"user@flink.apache.org" 
> *Date: *Friday, October 14, 2016 at 6:57 PM
> *To: *"user@flink.apache.org" 
> *Subject: *Re: job failure with checkpointing enabled
>
>
>
> Hi,
>
> the file that Flink is trying to create there is not meant to be in the
> checkpointing location. It is a local file that is used for buffering
> elements until a checkpoint barrier arrives (for certain cases). Can you
> check whether the base path where it is trying to create that file exists?
> For the exception that you posted that would be:
> /tmp/flink-io-202fdf67-3f8c-47dd-8ebc-2265430644ed
>
>
>
> Cheers,
>
> Aljoscha
>
>
>
> On Fri, 14 Oct 2016 at 17:37  wrote:
>
> I recently tried enabling checkpointing in a job (that previously works
> w/o checkpointing) and received the following failure on job execution:
>
>
>
> java.io.FileNotFoundException:
> /tmp/flink-io-202fdf67-3f8c-47dd-8ebc-2265430644ed/a426eb27761575b3b79e464719bba96e16a1869d85bae292a2ef7eb72fa8a14c.0.buffer
> (No such file or directory)
>
> at java.io.RandomAccessFile.open0(Native Method)
>
> at java.io.RandomAccessFile.open(RandomAccessFile.java:316)
>
> at java.io.RandomAccessFile.(RandomAccessFile.java:243)
>
> at
> org.apache.flink.streaming.runtime.io.BufferSpiller.createSpillingChannel(BufferSpiller.java:247)
>
> at
> org.apache.flink.streaming.runtime.io.BufferSpiller.(BufferSpiller.java:117)
>
> at
> org.apache.flink.streaming.runtime.io.BarrierBuffer.(BarrierBuffer.java:94)
>
> at
> org.apache.flink.streaming.runtime.io.StreamInputProcessor.(StreamInputProcessor.java:96)
>
> at
> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.init(OneInputStreamTask.java:49)
>
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:239)
>
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
>
> at java.lang.Thread.run(Thread.java:745)
>
>
>
>
>
> The job then restarts and fails again in an endless cycle.
>
>
>
> This feels like a configuration issue.  My guess is that Flink is looking
> for the file above on local storage, though we’ve configured checkpointing
> to use hdfs (see below).
>
>
>
> To enable checkpointing, this is what I did:
>
> env.enableCheckpointing(3000l);
>
>
>
> Relevant configurations in flink-conf.yaml:
>
> state.backend: filesystem
>
> state.backend.fs.checkpointdir:
> hdfs://myhadoopnamenode:8020/apps/flink/checkpoints
>
>
>
> Note, the directory we’ve configured is not the same as the path indicated
> in the error.
>
>
>
> Interestingly, there are plenty of subdirs in my checkpoints directory,
> these appear to correspond to job start times, even though these jobs don’t
> have checkpointing enabled:
>
> drwxr-xr-x   - rtap hdfs  0 2016-10-13 07:48
> /apps/flink/checkpoints/b4870565f148cff10478dca8bff27bf7
>
> drwxr-xr-x   - rtap hdfs  0 2016-10-13 08:27
> /apps/flink/checkpoints/044b21a0f252b6142e7ddfee7bfbd7d5
>
> drwxr-xr-x   - rtap hdfs  0 2016-10-13 08:36
> /apps/flink/checkpoints/a658b23c2d2adf982a2cf317bfb3d3de
>
> drwxr-xr-x   - rtap hdfs  0 2016-10-14 07:38
> /apps/flink/checkpoints/1156bd1796105ad95a8625cb28a0b816
>
> drwxr-xr-x   - rtap hdfs  0 2016-10-14 07:41
> /apps/flink/checkpoints/58fdd94b7836a3b3ed9abc5c8f3a1dd5
>
> drwxr-xr-x   - rtap hdfs  0 2016-10-14 07:43
> /apps/flink/checkpoints/47a849a8ed6538b9e7d3826a628d38b9
>
> drwxr-xr-x   - rtap hdfs  0 2016-10-14 07:49
> /apps/flink/checkpoints/e6a9e2300ea5c36341fa160adab789f0
>
>
>
> Thanks!
>
>
>
>
>
>
> --
>
> The information contained 

Testing a streaming data flow

[bart.kastermans]

What are the standard approaches for testing a streaming algorithm?  I have been
able to come up with the below where I

1) create a data source that emits events in bunches with set times so that I
know the events will be in the same window,
2) end the stream with a mapWithState where the state checks if the expected
elements pass by in the expected order.

This does not seem like the most robust way of doing this.  Suggestions?

Best,
Bart


import java.io.{FileWriter, StringWriter}
import java.util.{Calendar, Date, Properties}

import com.fasterxml.jackson.annotation.JsonIgnoreProperties
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.functions.source.SourceFunction
import 
org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer09, 
FlinkKafkaProducer09}
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import org.slf4j.{Logger, LoggerFactory}

import scala.math.max
import scala.util.{Random, hashing}
import java.time


object SessionizePageviewsTT {

  val logger: Logger = LoggerFactory.getLogger("SessionizePageviewsTT") // 
classOf doesn't work on an object

  def get_now_ms(): Long = {
System.currentTimeMillis()
  }

  def main(args: Array[String]) {

val env: StreamExecutionEnvironment = 
StreamExecutionEnvironment.getExecutionEnvironment

val xs = 1 to 30
def sourceFF(scx: SourceContext[Int]): Unit = {
  var cur = 1
  var now: Long = get_now_ms()
  while (cur < 31) {
// every 10 wait 10 seconds and then burst a bunch
if (cur % 10 == 0) {
  Thread.sleep(1)
  now = get_now_ms()
}
println("emiting: " + cur + ", " + now)
scx.collectWithTimestamp(cur, now)
cur += 1
  }
}

val x: DataStream[Int] = env.addSource(sourceFF _)

val vals = List(45, 145);

def checkFF(xy: (Int, Int), s: Option[Int]): ((Int, Int), Option[Int]) = {
  val idx = if (s.isDefined) s.get else 0
  if (idx < vals.length) {
if (xy._1 == vals(idx)) {
  println("all ok")
} else {
  println("error error")
}
  }

  (xy, Some(idx + 1))
}

x.map(x => (x,1)).keyBy(1).timeWindow(Time.seconds(10)).sum(0).keyBy(x => 
1).mapWithState(checkFF).print

env.execute(s"")
  }
}

Re: job failure with checkpointing enabled

[robert.lancaster]

HI Aljoscha,

Thanks for the response.

To answer your question, the base path did not exist.  But, I think I found the 
issue.  I believe I had some rogue task managers running.  As a troubleshooting 
step, I attempted to restart my cluster.  However, after shutting down the 
cluster I noticed that there were still task managers running on most of my 
nodes (and on the master).  Interestingly, on a second attempt to shut down the 
cluster, I received the message “No taskmanager daemon to stop on host…” for 
each of my nodes, even though I could see the flink processes running on these 
machines.   After manually killing these processes and restarting the cluster, 
the problem went away.

So, my assumption is that on a previous attempt to bounce the cluster, these 
processes did not shut down cleanly.  Starting the cluster after that *may* 
have resulted in second instances of the task manager running on most nodes.  
I’m not certain, however, and I haven’t yet been able to reproduce the issue.





From: Aljoscha Krettek 
Reply-To: "user@flink.apache.org" 
Date: Friday, October 14, 2016 at 6:57 PM
To: "user@flink.apache.org" 
Subject: Re: job failure with checkpointing enabled

Hi,
the file that Flink is trying to create there is not meant to be in the 
checkpointing location. It is a local file that is used for buffering elements 
until a checkpoint barrier arrives (for certain cases). Can you check whether 
the base path where it is trying to create that file exists? For the exception 
that you posted that would be: 
/tmp/flink-io-202fdf67-3f8c-47dd-8ebc-2265430644ed

Cheers,
Aljoscha

On Fri, 14 Oct 2016 at 17:37 
> wrote:
I recently tried enabling checkpointing in a job (that previously works w/o 
checkpointing) and received the following failure on job execution:

java.io.FileNotFoundException: 
/tmp/flink-io-202fdf67-3f8c-47dd-8ebc-2265430644ed/a426eb27761575b3b79e464719bba96e16a1869d85bae292a2ef7eb72fa8a14c.0.buffer
 (No such file or directory)
at java.io.RandomAccessFile.open0(Native Method)
at java.io.RandomAccessFile.open(RandomAccessFile.java:316)
at java.io.RandomAccessFile.(RandomAccessFile.java:243)
at 
org.apache.flink.streaming.runtime.io.BufferSpiller.createSpillingChannel(BufferSpiller.java:247)
at 
org.apache.flink.streaming.runtime.io.BufferSpiller.(BufferSpiller.java:117)
at 
org.apache.flink.streaming.runtime.io.BarrierBuffer.(BarrierBuffer.java:94)
at 
org.apache.flink.streaming.runtime.io.StreamInputProcessor.(StreamInputProcessor.java:96)
at 
org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.init(OneInputStreamTask.java:49)
at 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:239)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
at java.lang.Thread.run(Thread.java:745)


The job then restarts and fails again in an endless cycle.

This feels like a configuration issue.  My guess is that Flink is looking for 
the file above on local storage, though we’ve configured checkpointing to use 
hdfs (see below).

To enable checkpointing, this is what I did:
env.enableCheckpointing(3000l);

Relevant configurations in flink-conf.yaml:
state.backend: filesystem
state.backend.fs.checkpointdir: 
hdfs://myhadoopnamenode:8020/apps/flink/checkpoints

Note, the directory we’ve configured is not the same as the path indicated in 
the error.

Interestingly, there are plenty of subdirs in my checkpoints directory, these 
appear to correspond to job start times, even though these jobs don’t have 
checkpointing enabled:
drwxr-xr-x   - rtap hdfs  0 2016-10-13 07:48 
/apps/flink/checkpoints/b4870565f148cff10478dca8bff27bf7
drwxr-xr-x   - rtap hdfs  0 2016-10-13 08:27 
/apps/flink/checkpoints/044b21a0f252b6142e7ddfee7bfbd7d5
drwxr-xr-x   - rtap hdfs  0 2016-10-13 08:36 
/apps/flink/checkpoints/a658b23c2d2adf982a2cf317bfb3d3de
drwxr-xr-x   - rtap hdfs  0 2016-10-14 07:38 
/apps/flink/checkpoints/1156bd1796105ad95a8625cb28a0b816
drwxr-xr-x   - rtap hdfs  0 2016-10-14 07:41 
/apps/flink/checkpoints/58fdd94b7836a3b3ed9abc5c8f3a1dd5
drwxr-xr-x   - rtap hdfs  0 2016-10-14 07:43 
/apps/flink/checkpoints/47a849a8ed6538b9e7d3826a628d38b9
drwxr-xr-x   - rtap hdfs  0 2016-10-14 07:49 
/apps/flink/checkpoints/e6a9e2300ea5c36341fa160adab789f0

Thanks!




The information contained in this communication is confidential and intended 
only for the use of the recipient named above, and may be legally privileged 
and exempt from disclosure under applicable law. If the reader of this message 
is not the intended recipient, you are hereby notified that any dissemination, 
distribution or copying of this communication is strictly prohibited. If you 
have received this 

Re: "java.net.SocketException: No buffer space available (maximum connections reached?)" when reading from HDFS

[Stephan Ewen]

Happy to hear it!

On Mon, Oct 17, 2016 at 9:31 AM, Yassine MARZOUGUI <
y.marzou...@mindlytix.com> wrote:

> That solved my problem, Thank you!
>
> Best,
> Yassine
>
> 2016-10-16 19:18 GMT+02:00 Stephan Ewen :
>
>> Hi!
>>
>> Looks to me that this is the following problem: The Decompression Streams
>> did not properly forward the "close()" calls.
>>
>> It is in the lastest 1.2-SNAPSHOT, but did not make it into version
>> 1.1.3.
>> The fix is in that pull request: https://github.com/apache/flin
>> k/pull/2581
>>
>> I have pushed the fix into the latest 1.1-SNAPSHOT branch.
>>
>> If you get the code via "git clone -b release-1.1
>> https://github.com/apache/flink.git; you will get the code that is the
>> same as the 1.1.3 release, plus the patch to this problem.
>>
>> Greetings,
>> Stephan
>>
>>
>> On Sat, Oct 15, 2016 at 10:11 PM, Yassine MARZOUGUI <
>> y.marzou...@mindlytix.com> wrote:
>>
>>> Hi all,
>>>
>>> I'm reading a large number of small files from HDFS in batch mode (about
>>> 20 directories, each directory contains about 3000 files, using
>>> recursive.file.enumeration=true), and each time, at about 200 GB of
>>> received data, my job fails with the following exception:
>>>
>>> java.io.IOException: Error opening the Input Split
>>> hdfs:///filepath/filename.csv.gz [0,-1]: Could not obtain block:
>>> BP-812793611-127.0.0.1-1455882335652:blk_1075977174_2237313
>>> file=/filepath/filename.csv.gz
>>> at org.apache.flink.api.common.io.FileInputFormat.open(FileInpu
>>> tFormat.java:693)
>>> at org.apache.flink.api.common.io.DelimitedInputFormat.open(Del
>>> imitedInputFormat.java:424)
>>> at org.apache.flink.api.common.io.DelimitedInputFormat.open(Del
>>> imitedInputFormat.java:47)
>>> at org.apache.flink.runtime.operators.DataSourceTask.invoke(Dat
>>> aSourceTask.java:140)
>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
>>> at java.lang.Thread.run(Unknown Source)
>>> Caused by: org.apache.hadoop.hdfs.BlockMissingException: Could not
>>> obtain block: BP-812793611-127.0.0.1-1455882335652:blk_1075977174_2237313
>>> file=/filepath/filename.csv.gz
>>> at org.apache.hadoop.hdfs.DFSInputStream.chooseDataNode(DFSInpu
>>> tStream.java:984)
>>> at org.apache.hadoop.hdfs.DFSInputStream.blockSeekTo(DFSInputSt
>>> ream.java:642)
>>> at org.apache.hadoop.hdfs.DFSInputStream.readWithStrategy(DFSIn
>>> putStream.java:882)
>>> at org.apache.hadoop.hdfs.DFSInputStream.read(DFSInputStream.ja
>>> va:934)
>>> at org.apache.hadoop.hdfs.DFSInputStream.read(DFSInputStream.ja
>>> va:735)
>>> at java.io.FilterInputStream.read(Unknown Source)
>>> at org.apache.flink.runtime.fs.hdfs.HadoopDataInputStream.read(
>>> HadoopDataInputStream.java:59)
>>> at java.util.zip.CheckedInputStream.read(Unknown Source)
>>> at java.util.zip.GZIPInputStream.readUByte(Unknown Source)
>>> at java.util.zip.GZIPInputStream.readUShort(Unknown Source)
>>> at java.util.zip.GZIPInputStream.readHeader(Unknown Source)
>>> at java.util.zip.GZIPInputStream.(Unknown Source)
>>> at java.util.zip.GZIPInputStream.(Unknown Source)
>>> at org.apache.flink.api.common.io.compression.GzipInflaterInput
>>> StreamFactory.create(GzipInflaterInputStreamFactory.java:44)
>>> at org.apache.flink.api.common.io.compression.GzipInflaterInput
>>> StreamFactory.create(GzipInflaterInputStreamFactory.java:31)
>>> at org.apache.flink.api.common.io.FileInputFormat.decorateInput
>>> Stream(FileInputFormat.java:717)
>>> at org.apache.flink.api.common.io.FileInputFormat.open(FileInpu
>>> tFormat.java:689)
>>> ... 5 more
>>>
>>> I checked the file each time and it exists and is healthy. Looking at
>>> the taskmanager logs, I found the following exceptions which suggests it is
>>> running out of connections:
>>>
>>> 2016-10-15 18:20:27,034 WARN  org.apache.hadoop.hdfs.BlockReaderFactory
>>> - I/O error constructing remote block reader.
>>> java.net.SocketException: No buffer space available (maximum connections
>>> reached?): connect
>>> at sun.nio.ch.Net.connect0(Native Method)
>>> at sun.nio.ch.Net.connect(Unknown Source)
>>> at sun.nio.ch.Net.connect(Unknown Source)
>>> at sun.nio.ch.SocketChannelImpl.connect(Unknown Source)
>>> at org.apache.hadoop.net.SocketIOWithTimeout.connect(SocketIOWi
>>> thTimeout.java:192)
>>> at org.apache.hadoop.net.NetUtils.connect(NetUtils.java:531)
>>> at org.apache.hadoop.hdfs.DFSClient.newConnectedPeer(DFSClient.
>>> java:3436)
>>> at org.apache.hadoop.hdfs.BlockReaderFactory.nextTcpPeer(BlockR
>>> eaderFactory.java:777)
>>> at org.apache.hadoop.hdfs.BlockReaderFactory.getRemoteBlockRead
>>> erFromTcp(BlockReaderFactory.java:694)
>>> at org.apache.hadoop.hdfs.BlockReaderFactory.build(BlockReaderF
>>> actory.java:355)
>>> at org.apache.hadoop.hdfs.DFSInputStream.blockSeekTo(DFSInputSt
>>> ream.java:673)
>>> 

Re: how can I name a sink?

[Márton Balassi]

No problem, it is great that you have found the solution.

On Mon, Oct 17, 2016 at 12:16 PM, 侯林蔚  wrote:

> More information.
>   my code are like this:
> [image: 内嵌图片 1]
>
> and I find I can name a sink by change code like this :
>
> [image: 内嵌图片 2]
>
> sorry for my reckless behavior.
>
> 2016-10-17 17:48 GMT+08:00 侯林蔚 :
>
>> hi
>>I make a flink topology and run it on my dev-cluster.
>>but I find something on the picture as follow:
>>[image: 内嵌图片 1]
>>
>>all my sinks are unnamed , is there any method to name a sink?
>>
>>
>>   thank you very much.
>>
>
>

Re: how can I name a sink?

[侯林蔚]

More information.
  my code are like this:
[image: 内嵌图片 1]

and I find I can name a sink by change code like this :

[image: 内嵌图片 2]

sorry for my reckless behavior.

2016-10-17 17:48 GMT+08:00 侯林蔚 :

> hi
>I make a flink topology and run it on my dev-cluster.
>but I find something on the picture as follow:
>[image: 内嵌图片 1]
>
>all my sinks are unnamed , is there any method to name a sink?
>
>
>   thank you very much.
>

BoundedOutOfOrdernessTimestampExtractor and allowedlateness

[Yassine MARZOUGUI]

Hi,

I'm a bit confused about how Flink deals with late elements after the
introduction of allowedlateness to windows. What is the difference between
using a BoundedOutOfOrdernessTimestampExtractor(Time.seconds(X)) and
allowedlateness(Time.seconds(X))? What if one is used and the other is not?
and what if a different lateness is used in each one? Could you please
clarify it on basis of a simple example? Thank you.

Best,
Yassine

Re: Flink Metrics

[Chesnay Schepler]

Hello,

we could also offer a small utility method that creates 3 flink meters, 
each reporting one rate of a DW meter.


Timers weren't added yet since, as Till said, no one requested them yet 
and we haven't found a proper internal use-case for them


Regards,
Chesnay

On 17.10.2016 09:52, Till Rohrmann wrote:


Hi Govind,

I think the |DropwizardMeterWrapper| implementation is just a 
reference implementation where it was decided to report the minute 
rate. You can define your own meter class which allows to configure 
the rate interval accordingly.


Concerning Timers, I think nobody requested this metric so far. If you 
want, then you can open a JIRA issue and contribute it. The community 
would really appreciate that.


Cheers,
Till

​

On Mon, Oct 17, 2016 at 5:26 AM, Govindarajan Srinivasaraghavan 
> wrote:


Hi,

I am currently using flink 1.2 snapshot and instrumenting my
pipeline with flink metrics. One small suggestion I have is
currently the Meter interface only supports getRate() which is
always the one minute rate.

It would great if all the rates (1 min, 5 min & 15 min) are
exposed to get a better picture in terms of performance.

Also is there any reason why timers are not part of flink metrics
core?

Regards,
Govind

Re: First Program with WordCount - Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/flink/api/common/functions/FlatMapFunction

[Stefan Richter]

Hi,

looks like there is no Flink jar in the classpath with which you run your 
program. You need to make sure that they relevant jars are there or else your 
program cannot find Flink’s classes, leading to a ClassNotFoundException.

Best,
Stefan

> Am 16.10.2016 um 19:26 schrieb Kaepke, Marc :
> 
> Hi guys,
> 
> I followed this guide 
> (https://ci.apache.org/projects/flink/flink-docs-release-1.2/quickstart/java_api_quickstart.html
>  
> ),
>  but I get an Exception if I run WordCount 
> 
> /usr/lib/jvm/java-8-oracle/bin/java -Didea.launcher.port=7536 
> -Didea.launcher.bin.path=/home/marc/Programs/idea-IC-162.2032.8/bin 
> -Dfile.encoding=UTF-8 -classpath 
> "/usr/lib/jvm/java-8-oracle/jre/lib/charsets.jar:/usr/lib/jvm/java-8-oracle/jre/lib/deploy.jar:/usr/lib/jvm/java-8-oracle/jre/lib/ext/cldrdata.jar:/usr/lib/jvm/java-8-oracle/jre/lib/ext/dnsns.jar:/usr/lib/jvm/java-8-oracle/jre/lib/ext/jaccess.jar:/usr/lib/jvm/java-8-oracle/jre/lib/ext/jfxrt.jar:/usr/lib/jvm/java-8-oracle/jre/lib/ext/localedata.jar:/usr/lib/jvm/java-8-oracle/jre/lib/ext/nashorn.jar:/usr/lib/jvm/java-8-oracle/jre/lib/ext/sunec.jar:/usr/lib/jvm/java-8-oracle/jre/lib/ext/sunjce_provider.jar:/usr/lib/jvm/java-8-oracle/jre/lib/ext/sunpkcs11.jar:/usr/lib/jvm/java-8-oracle/jre/lib/ext/zipfs.jar:/usr/lib/jvm/java-8-oracle/jre/lib/javaws.jar:/usr/lib/jvm/java-8-oracle/jre/lib/jce.jar:/usr/lib/jvm/java-8-oracle/jre/lib/jfr.jar:/usr/lib/jvm/java-8-oracle/jre/lib/jfxswt.jar:/usr/lib/jvm/java-8-oracle/jre/lib/jsse.jar:/usr/lib/jvm/java-8-oracle/jre/lib/management-agent.jar:/usr/lib/jvm/java-8-oracle/jre/lib/plugin.jar:/usr/lib/jvm/java-8-oracle/jre/lib/resources.jar:/usr/lib/jvm/java-8-oracle/jre/lib/rt.jar:/home/marc/apache
>  
> flink/flink.gelly/target/classes:/home/marc/Programs/idea-IC-162.2032.8/lib/idea_rt.jar"
>  com.intellij.rt.execution.application.AppMain 
> haw.bachelor.flink.gelly.WordCount
> Exception in thread "main" java.lang.NoClassDefFoundError: 
> org/apache/flink/api/common/functions/FlatMapFunction
> at java.lang.Class.forName0(Native Method)
> at java.lang.Class.forName(Class.java:264)
> at com.intellij.rt.execution.application.AppMain.main(AppMain.java:123)
> Caused by: java.lang.ClassNotFoundException: 
> org.apache.flink.api.common.functions.FlatMapFunction
> at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> ... 3 more
> 
> 
> my environment:
> Ubuntu 14.04 LTS
> Oracle Java 8
> Maven 3.0.5
> intellij communityedition
> 
> 
> thanks for help
> Marc

Re: Flink Metrics

[Till Rohrmann]

Hi Govind,

I think the DropwizardMeterWrapper implementation is just a reference
implementation where it was decided to report the minute rate. You can
define your own meter class which allows to configure the rate interval
accordingly.

Concerning Timers, I think nobody requested this metric so far. If you
want, then you can open a JIRA issue and contribute it. The community would
really appreciate that.

Cheers,
Till
​

On Mon, Oct 17, 2016 at 5:26 AM, Govindarajan Srinivasaraghavan <
govindragh...@gmail.com> wrote:

> Hi,
>
> I am currently using flink 1.2 snapshot and instrumenting my pipeline with
> flink metrics. One small suggestion I have is currently the Meter interface
> only supports getRate() which is always the one minute rate.
>
> It would great if all the rates (1 min, 5 min & 15 min) are exposed to get
> a better picture in terms of performance.
>
> Also is there any reason why timers are not part of flink metrics core?
>
> Regards,
> Govind
>

Re: "java.net.SocketException: No buffer space available (maximum connections reached?)" when reading from HDFS

[Yassine MARZOUGUI]

That solved my problem, Thank you!

Best,
Yassine

2016-10-16 19:18 GMT+02:00 Stephan Ewen :

> Hi!
>
> Looks to me that this is the following problem: The Decompression Streams
> did not properly forward the "close()" calls.
>
> It is in the lastest 1.2-SNAPSHOT, but did not make it into version 1.1.3.
> The fix is in that pull request: https://github.com/apache/flink/pull/2581
>
> I have pushed the fix into the latest 1.1-SNAPSHOT branch.
>
> If you get the code via "git clone -b release-1.1
> https://github.com/apache/flink.git; you will get the code that is the
> same as the 1.1.3 release, plus the patch to this problem.
>
> Greetings,
> Stephan
>
>
> On Sat, Oct 15, 2016 at 10:11 PM, Yassine MARZOUGUI <
> y.marzou...@mindlytix.com> wrote:
>
>> Hi all,
>>
>> I'm reading a large number of small files from HDFS in batch mode (about
>> 20 directories, each directory contains about 3000 files, using
>> recursive.file.enumeration=true), and each time, at about 200 GB of
>> received data, my job fails with the following exception:
>>
>> java.io.IOException: Error opening the Input Split
>> hdfs:///filepath/filename.csv.gz [0,-1]: Could not obtain block:
>> BP-812793611-127.0.0.1-1455882335652:blk_1075977174_2237313
>> file=/filepath/filename.csv.gz
>> at org.apache.flink.api.common.io.FileInputFormat.open(FileInpu
>> tFormat.java:693)
>> at org.apache.flink.api.common.io.DelimitedInputFormat.open(Del
>> imitedInputFormat.java:424)
>> at org.apache.flink.api.common.io.DelimitedInputFormat.open(Del
>> imitedInputFormat.java:47)
>> at org.apache.flink.runtime.operators.DataSourceTask.invoke(
>> DataSourceTask.java:140)
>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
>> at java.lang.Thread.run(Unknown Source)
>> Caused by: org.apache.hadoop.hdfs.BlockMissingException: Could not
>> obtain block: BP-812793611-127.0.0.1-1455882335652:blk_1075977174_2237313
>> file=/filepath/filename.csv.gz
>> at org.apache.hadoop.hdfs.DFSInputStream.chooseDataNode(DFSInpu
>> tStream.java:984)
>> at org.apache.hadoop.hdfs.DFSInputStream.blockSeekTo(DFSInputSt
>> ream.java:642)
>> at org.apache.hadoop.hdfs.DFSInputStream.readWithStrategy(DFSIn
>> putStream.java:882)
>> at org.apache.hadoop.hdfs.DFSInputStream.read(DFSInputStream.
>> java:934)
>> at org.apache.hadoop.hdfs.DFSInputStream.read(DFSInputStream.
>> java:735)
>> at java.io.FilterInputStream.read(Unknown Source)
>> at org.apache.flink.runtime.fs.hdfs.HadoopDataInputStream.read(
>> HadoopDataInputStream.java:59)
>> at java.util.zip.CheckedInputStream.read(Unknown Source)
>> at java.util.zip.GZIPInputStream.readUByte(Unknown Source)
>> at java.util.zip.GZIPInputStream.readUShort(Unknown Source)
>> at java.util.zip.GZIPInputStream.readHeader(Unknown Source)
>> at java.util.zip.GZIPInputStream.(Unknown Source)
>> at java.util.zip.GZIPInputStream.(Unknown Source)
>> at org.apache.flink.api.common.io.compression.GzipInflaterInput
>> StreamFactory.create(GzipInflaterInputStreamFactory.java:44)
>> at org.apache.flink.api.common.io.compression.GzipInflaterInput
>> StreamFactory.create(GzipInflaterInputStreamFactory.java:31)
>> at org.apache.flink.api.common.io.FileInputFormat.decorateInput
>> Stream(FileInputFormat.java:717)
>> at org.apache.flink.api.common.io.FileInputFormat.open(FileInpu
>> tFormat.java:689)
>> ... 5 more
>>
>> I checked the file each time and it exists and is healthy. Looking at the
>> taskmanager logs, I found the following exceptions which suggests it is
>> running out of connections:
>>
>> 2016-10-15 18:20:27,034 WARN  org.apache.hadoop.hdfs.BlockReaderFactory
>> - I/O error constructing remote block reader.
>> java.net.SocketException: No buffer space available (maximum connections
>> reached?): connect
>> at sun.nio.ch.Net.connect0(Native Method)
>> at sun.nio.ch.Net.connect(Unknown Source)
>> at sun.nio.ch.Net.connect(Unknown Source)
>> at sun.nio.ch.SocketChannelImpl.connect(Unknown Source)
>> at org.apache.hadoop.net.SocketIOWithTimeout.connect(SocketIOWi
>> thTimeout.java:192)
>> at org.apache.hadoop.net.NetUtils.connect(NetUtils.java:531)
>> at org.apache.hadoop.hdfs.DFSClient.newConnectedPeer(DFSClient.java:3436)
>> at org.apache.hadoop.hdfs.BlockReaderFactory.nextTcpPeer(BlockR
>> eaderFactory.java:777)
>> at org.apache.hadoop.hdfs.BlockReaderFactory.getRemoteBlockRead
>> erFromTcp(BlockReaderFactory.java:694)
>> at org.apache.hadoop.hdfs.BlockReaderFactory.build(BlockReaderF
>> actory.java:355)
>> at org.apache.hadoop.hdfs.DFSInputStream.blockSeekTo(DFSInputSt
>> ream.java:673)
>> at org.apache.hadoop.hdfs.DFSInputStream.readWithStrategy(DFSIn
>> putStream.java:882)
>> at org.apache.hadoop.hdfs.DFSInputStream.read(DFSInputStream.java:934)
>> at