Re: How to prepare data for K means clustering

[Chiwan Park]

Hi Ashutosh,

You can use basic Flink DataSet operations such as map and filter to transform 
your data. Basically, you have to declare a distance metric between each record 
in data. In example, we use euclidean distance (see euclideanDistance method in 
Point class).

In map method in SelectNearestCenter class, euclideanDistance method is used to 
measure the distance between each point. For your implementation, you have to 
substitute type to your data type (It can be your custom class or 
Flink-provided Tuple) and change distance metric for your data.

Regards,
Chiwan Park

> On Jan 21, 2016, at 4:14 PM, Ashutosh Kumar  
> wrote:
> 
> I saw example code for K means clustering . It takes input  data points as 
> pair of double values (1.2 2.3\n5.3 7.2\.). My question is how do I convert 
> my business data to this format. I have customer data which has columns like 
> house hold income , education and several others. I want to do clustering on 
> multiple columns something like Neilsen segments. 
> 
> Thanks
> Ashutosh

How to prepare data for K means clustering

[Ashutosh Kumar]

I saw example code for K means clustering . It takes input  data points as
pair of double values (1.2 2.3\n5.3 7.2\.). My question is how do I convert
my business data to this format. I have customer data which has columns
like house hold income , education and several others. I want to do
clustering on multiple columns something like Neilsen segments.

Thanks
Ashutosh

Re: Compile fails with scala 2.11.4

[Chiwan Park]

Thanks for sharing, Ritesh!

Regards,
Chiwan Park

> On Jan 21, 2016, at 12:28 AM, Ritesh Kumar Singh 
>  wrote:
> 
> Thanks for the update Robert, I tried it out and it works fine for 
> scala_2.11.4 version.
> I've made a docker image of the same and put it up on the hub just in case 
> anyone else needs it.
> 
> Thanks,
> 
> Ritesh Kumar Singh,
> https://riteshtoday.wordpress.com/
> 
> On Wed, Jan 20, 2016 at 10:07 AM, Robert Metzger  wrote:
> Hi,
> 
> in the latest master, the "tools/change-scala-version.sh" should be fixed. 
> Also, the 1.0-SNAPSHOT version deployed to the snapshot repository should be 
> good again.
> 
> @Ritesh: The commands were correct. I'm not sure if Flink builds with Scala 
> 2.11.4, the default 2.11 version we are using is 2.11.7.
> 
> 
> 
> On Tue, Jan 19, 2016 at 7:41 AM, Prez Cannady  
> wrote:
> Assuming you haven’t already migrated back to 2.10, you might try this;
> 
> $ git checkout release-0.10
> $ tools/change-scala-version 2.11
> $ mvn clean install -DskipTests=true -Dmaven.javadoc.skip=true 
> -Dscala.version=2.11.4 -Dscala.binary.version=2.11
> 
> Then try building your project.  Building under these conditions resolved my 
> issue which emitted the same error.
> 
> Prez Cannady  
> p: 617 500 3378  
> e: revp...@opencorrelate.org  
> GH: https://github.com/opencorrelate  
> LI: https://www.linkedin.com/in/revprez  
> 
> 
> 
> 
> 
> 
> 
> 
> 
>> On Jan 18, 2016, at 10:20 AM, Ritesh Kumar Singh 
>>  wrote:
>> 
>> Thanks for the replies.
>> 
>> @Chiwan, I am switching back to scala_2.10.4 for the time being. I was using 
>> scala_2.11.4 as this is the version I've compiled spark with. But anyways, I 
>> can wait for the bug to be resolved.
>> 
>> @Robert, the commands were as follows:
>> $tools/change-scala-version.sh 2.11
>> $mvn clean install -DskipTests -Dscala.version=2.11.4
>> 
>> I hope I'm doing it right ?
>> 
>> Thanks,
>> 
>> Ritesh Kumar Singh,
>> https://riteshtoday.wordpress.com/
>> 
>> On Mon, Jan 18, 2016 at 12:03 PM, Robert Metzger  wrote:
>> How did start the Flink for Scala 2.11 compilation ?
>> 
>> On Mon, Jan 18, 2016 at 11:41 AM, Chiwan Park  wrote:
>> Hi Ritesh,
>> 
>> This problem seems already reported [1]. Flink community is investigating 
>> this issue. I think that if you don’t need Scala 2.11, use Scala 2.10 until 
>> the issue is solved.
>> 
>> [1]: 
>> http://mail-archives.apache.org/mod_mbox/flink-user/201601.mbox/%3CCAB6CeiZ_2snN-piXzd3gHnyQePu_PA0Ro7qXUF8%3DVTxoyL0YyA%40mail.gmail.com%3E
>> 
>> > On Jan 18, 2016, at 7:24 PM, Ritesh Kumar Singh 
>> >  wrote:
>> >
>> > [ERROR] 
>> > /home/flink/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala:703:
>> >  error: can't expand macros compiled by previous versions of Scala
>> > [ERROR]   assert(cachedGraph2.isArchived)
>> > [ERROR]   ^
>> > [ERROR] one error found
>> > [INFO] 
>> > 
>> > [INFO] Reactor Summary:
>> > [INFO]
>> > [INFO] flink .. SUCCESS [ 
>> > 24.820 s]
>> > [INFO] flink-annotations .. SUCCESS [  
>> > 2.755 s]
>> > [INFO] flink-shaded-hadoop  SUCCESS [  
>> > 0.208 s]
>> > [INFO] flink-shaded-hadoop2 ... SUCCESS [ 
>> > 15.627 s]
>> > [INFO] flink-shaded-include-yarn-tests  SUCCESS [ 
>> > 17.076 s]
>> > [INFO] flink-shaded-curator ... SUCCESS [  
>> > 0.200 s]
>> > [INFO] flink-shaded-curator-recipes ... SUCCESS [  
>> > 2.751 s]
>> > [INFO] flink-shaded-curator-test .. SUCCESS [  
>> > 0.355 s]
>> > [INFO] flink-core . SUCCESS [ 
>> > 33.052 s]
>> > [INFO] flink-java . SUCCESS [ 
>> > 10.224 s]
>> > [INFO] flink-runtime .. FAILURE [01:23 
>> > min]
>> > [INFO] flink-optimizer  SKIPPED
>> >
>> >
>> > Any workaround for scala_2.11.4 or do I have to switch back to 
>> > scala_2.10.4 ?
>> >
>> > Thanks,
>> > Ritesh Kumar Singh,
>> > https://riteshtoday.wordpress.com/
>> >
>> 
>> Regards,
>> Chiwan Park
>> 
>> 
>> 
>> 
> 
> 
> 
> 

Re: integration with a scheduler

[Serkan Taş]

Thank you very much Stephan and Robert.

As Robert offers, the most common way is to execute a batch script, but i want 
to go beyond.

i am going to work on both alternatives.

Best regards,


> 20 Oca 2016 tarihinde 22:53 saatinde, Stephan Ewen  şunları 
> yazdı:
> 
> If you want to programmatically start / stop / cancel jobs, have a look at 
> the class 
> 
> "Client" 
> (https://github.com/apache/flink/blob/master/flink-clients/src/main/java/org/apache/flink/client/program/Client.java
>  
> )
> 
> From the classes RemoteEnvironment or the RemoteExecutor, you can see how to 
> use it to send jobs and commands to a Flink cluster.
> 
> 
> 
> 
> On Wed, Jan 20, 2016 at 3:03 PM, Robert Metzger  > wrote:
> Hi Serkan,
> 
> I would suggest to have a look at the "./bin/flink" tool. It allows you to 
> start ("run") and stop ("cancel") batch and streaming jobs.
> Flink doesn't support suspending jobs.
> 
> You can also use the JobManager web interface (default port: 8081) to get the 
> status of the job and also to cancel it.
> 
> If you need programmatic access to the job status, have a look at the 
> CliFrontend class. Its the class called by the "./bin/flink" tool. There, you 
> can see how jobs are submitted and cancelled.
> 
> Regards,
> Robert
> 
> 
> 2016-01-19 12:10 GMT+01:00  >:
> Hi,
> 
> I am planning to integrate flink with our job scheduler product to execute
> jobs - especially bathc like - on flink which may be the part of some
> other DAG style job chain.
> 
> I need some control ablities like start, stop, suspend, get status...
> 
> Where shold i go through ?
> 
> --
> Serkan Tas
> Likya Bilgi Teknolojileri
> ve Iletişim Hiz. Ltd.
> www.likyateknoloji.com 
> Tel : 0 216 471 81 55 
> Gsm : 0 542 242 00 92 
> Faks:  0 216 661 14 92 
> 
> --
> Bu elektronik posta ve onunla iletilen bütün dosyalar gizlidir. Sadece
> yukarıda isimleri belirtilen kişiler arasında özel haberleşme amacını
> taşımaktadır. Size yanlışlıkla ulaşmışsa bu elektonik postanın içeriğini
> açıklamanız, kopyalamanız, yönlendirmeniz ve kullanmanız kesinlikle
> yasaktır. Lütfen mesajı geri gönderiniz ve sisteminizden siliniz. Likya
> Bilgi Teknolojileri ve İletişim Hiz. Ltd. Şti. bu mesajın içeriği ile
> ilgili olarak hiç bir hukuksal sorumluluğu kabul etmez.
> 
> This electonic mail and any files transmitted with it are intended for the
> private use of  the persons named above. If you received this message in
> error, forwarding, copying or use of any of the information is strictly
> prohibited. Please immediately notify the sender and delete it from your
> system. Likya Bilgi Teknolojileri ve İletişim Hiz. Ltd. Şti. does not
> accept legal responsibility for the contents of this message.
> --
> 
> 
> Bu e-postayı yazdırmadan önce, çevreye olan sorumluluğunuzu tekrar düşünün.
> Please consider your environmental responsibility before printing this
> e-mail.
> 
> 
> 
> 
> 



Serkan Taş
Mobil : +90 532 250 07 71
Likya Bilgi Teknolojileri
ve İletişim Hiz. Ltd. Şti.
www.likyateknoloji.com 
 
--
Bu elektronik posta ve onunla iletilen bütün dosyalar gizlidir. Sadece yukarıda 
isimleri belirtilen kişiler arasında özel haberleşme amacını taşımaktadır. Size 
yanlışlıkla ulaşmışsa bu elektonik postanın içeriğini açıklamanız, 
kopyalamanız, yönlendirmeniz ve kullanmanız kesinlikle yasaktır. Lütfen mesajı 
geri gönderiniz ve sisteminizden siliniz. Likya Bilgi Teknolojileri ve İletişim 
Hiz. Ltd. Şti. bu mesajın içeriği ile ilgili olarak hiç bir hukuksal 
sorumluluğu kabul etmez.
 
This electronic mail and any files transmitted with it are intended for the 
private use of  the persons named above. If you received this message in error, 
forwarding, copying or use of any of the information is strictly prohibited. 
Please immediately notify the sender and delete it from your system. Likya 
Bilgi Teknolojileri ve İletişim Hiz. Ltd. Şti. does not accept legal 
responsibility for the contents of this message.
--








P
Bu e-postayı yazdırmadan önce, çevreye olan sorumluluğunuzu tekrar düşünün.
Please consider your environmental responsibility before printing this e-mail.
 

Re: InvalidTypesException - Input mismatch: Basic type 'Integer' expected but was 'Long'

[Stephan Ewen]

Hi!

Can you check if the problem persists in the 1.0-SNAPSHOT branch? It may be
fixed in the newest version already, since we cannot reproduce it the
latest version.

Thanks,
Staphen


On Wed, Jan 20, 2016 at 9:56 PM, Biplob Biswas 
wrote:

> Hello everyone,
>
> I am still stuck with this issue, can anyone point me in the right
> direction?
>
> Thanks & Regards
> Biplob Biswas
>
> On Mon, Jan 18, 2016 at 2:24 PM, Biplob Biswas 
> wrote:
>
>> Hi Till,
>>
>> I am using flink 0.10.1 and if i am not wrong it corresponds to the
>> 1.0-Snapshot you mentioned.
>>
>> [image: Inline image 1]
>>
>> If wrong, please suggest what should I do to fix it.
>>
>> Thanks & Regards
>> Biplob Biswas
>>
>> On Mon, Jan 18, 2016 at 11:23 AM, Till Rohrmann 
>> wrote:
>>
>>> Hi Biplob,
>>>
>>> which version of Flink are you using? With version 1.0-SNAPSHOT, I
>>> cannot reproduce your problem.
>>>
>>> Cheers,
>>> Till
>>> ​
>>>
>>> On Sun, Jan 17, 2016 at 4:56 PM, Biplob Biswas >> > wrote:
>>>
 Hi,

 I am getting the following exception when i am using the map function

 Exception in thread "main"
> org.apache.flink.api.common.functions.InvalidTypesException: The return
> type of function 'computeWeightedDistribution(GraphWeighted.java:73)' 
> could
> not be determined automatically, due to type erasure. You can give type
> information hints by using the returns(...) method on the result of the
> transformation call, or by letting your function implement the
> 'ResultTypeQueryable' interface.
> at org.apache.flink.api.java.DataSet.getType(DataSet.java:176)
> at org.apache.flink.api.java.DataSet.groupBy(DataSet.java:692)
> at
> aim3.GraphWeighted.computeWeightedDistribution(GraphWeighted.java:74)
> at aim3.SlashdotZooInDegree.main(SlashdotZooInDegree.java:39)
> Caused by:
> org.apache.flink.api.common.functions.InvalidTypesException: Input
> mismatch: Basic type 'Integer' expected but was 'Long'.
> at
> org.apache.flink.api.java.typeutils.TypeExtractor.validateInputType(TypeExtractor.java:767)
> at
> org.apache.flink.api.java.typeutils.TypeExtractor.getUnaryOperatorReturnType(TypeExtractor.java:276)
> at
> org.apache.flink.api.java.typeutils.TypeExtractor.getMapReturnTypes(TypeExtractor.java:110)
> at org.apache.flink.api.java.DataSet.map(DataSet.java:213)
> at
> aim3.GraphWeighted.computeWeightedDistribution(GraphWeighted.java:73)
> ... 1 more



 This is the part of the code which I am trying to run :

 DataSet> distinctVertex = sourceVertex
>  .union(destinationVertex)
>  .groupBy(0)
>  .aggregate(Aggregations.SUM, 1);
> // Compute the degrees (degree, count)
>
>  DataSet> degreeCount = distinctVertex
>  .map(new DegreeMapper())
>  .groupBy(0)
>  .aggregate(Aggregations.SUM, 1);



 and the error I am getting is at this line *.map(new DegreeMapper())*

 Also, the degree mapper is a simply map function which emits the second
 column and 1 as follows:

>
> public static class DegreeMapper implements
> MapFunction, Tuple2> {
> private static final long serialVersionUID = 1L;
> public Tuple2 map(Tuple2 input) throws
> Exception {
> return new Tuple2(input.f1, 1);
> }
> }



 Now I am lost as to what I did wrong and why I am getting that error,
 any help would be appreciated.

 Thanks a lot.

 Thanks & Regards
 Biplob Biswas

>>>
>>>
>>
>

Re: Actual byte-streams in multiple-node pipelines

[Stephan Ewen]

This sounds quite feasible, actually, though it is a pretty unique use case.

Like Robert said, you can write map() and flatMap() function on byte[]
arrays. Make sure that the byte[] that the sources produce are not super
small and not too large (I would start with 1-4K or so).

You can control how data flows pretty well. It flows 1:1 from produce to
consumer, if you simply chain these function calls after another. To
balance bytes across receivers, use rebalance(), broadcast(), or
partitionCustom().

Streams maintain order of elements, unless steams get split/merged by
operations like rebalance(), partition() / broadcast / keyBy() or similar.

To union multiple streams with control over how the result stream gets
pieced together, I would try to connect streams and use a CoMapFunction /
CoFlatMapFunction to stitch the result stream together form the two input
streams.

To get exactly-once processing guarantees, activate checkpointing and use a
source that supports that. If you use a custom source, you may need a few
lines to integrate it with the checkpoint mechanism, but that is very
doable.

Hope that helps!

Greetings,
Stephan




On Wed, Jan 20, 2016 at 2:20 PM, Tal Maoz  wrote:

> Hey Robert,
>
> Thanks for responding!
>
> The latency I'm talking about would be no more than 1 second from input to
> output (meaning, bytes should flow immediately through the pipline and get
> to the other side after going through the processing). You can assume the
> processors have enough power to work in real-time.
>
> The processors would be, for the most part, running external processes
> (binary executables) and will feed them the incoming data, and then pass
> along their stdout to the next stage. Simply put, I have several existing
> 'blackbox' utilities that I need to run on the data in sequence and each of
> which is a CPU hog...
>
> Regarding fault tolerance, no data should be lost and each processor
> should get the data ONCE and in the correct order (when data is supposed to
> flow to the same processor). If a node crashes, a new one will take it's
> place and the data that was sent to the crashed node and was not processed
> should be sent to the new one, while the output should flow transparently
> to the next node as if no crashes happened. I know this is a very
> complicated demand but it is a must in my case.
>
> Finally, I'm talking about multiple pipelines running, where each node in
> a pipeline will be pre-configured before data starts flowing. Each pipeline
> will read data from a socket or from an MQ if such an MQ exists and is able
> to handle the load with the required low-latency. Each pipeline's source
> could be at the range of 45-600MB/s (this can't be split into multiple
> sources) and eventually, with enough resources and scaling, the system
> should support hundreds of such pipelines, each with it's own source! Also,
> at some point, 2 or more sources could be joined with some transformation
> into a single data stream... Assume the network fabric itself is capable of
> moving those amounts of data...
>
> If i use DataStream where i divide a single segment into very
> small buffers for low-latency, how can ensure that, on the one hand the
> data for a single segments flows entirely to the same processor while on
> the other, different segments can be balanced between several processors?
>
>
> Tal
>
> On Wed, Jan 20, 2016 at 3:02 PM, Robert Metzger 
> wrote:
>
>> Hi Tal,
>>
>> that sounds like an interesting use case. I think I need a bit more
>> details about your use case to see how it can be done with Flink.
>> You said you need low latency, what latency is acceptable for you?
>>
>> Also, I was wondering how are you going to feed the input data into
>> Flink? If the data is coming from multiple sources, maybe everything can be
>> done completely parallel.
>> Do you need any fault tolerance guarantees?
>>
>> You can use Flink's DataStream abstraction with different data types, and
>> you could create a DataStream. Flink would internally still send
>> multiple of those records in one buffer.
>> I think the more efficient approach is, as you suggested, to use a
>> DataStream of larger chunks.
>>
>> What kind of transformations are you planning to do on the stream?
>>
>> Regarding the amount of data we are talking about here: Flink is
>> certainly able to handle those loads. I recently did some tests with our
>> KafkaConsumer and I was able to read 390 megabytes/second on my laptop,
>> using a parallelism of one (so only one reading thread). My SSD has a read
>> rate of 530 MBs/. With sufficiently fast hardware, a few Flink TaskManagers
>> will be able to read 600MB/s.
>>
>>
>>
>> On Wed, Jan 20, 2016 at 1:39 PM, Ritesh Kumar Singh <
>> riteshoneinamill...@gmail.com> wrote:
>>
>>> I think with sufficient processing power flink can do the above
>>> mentioned task using the stream api
>>> 
>>> .
>>>
>>> Thanks,

Re: InvalidTypesException - Input mismatch: Basic type 'Integer' expected but was 'Long'

[Biplob Biswas]

Hello everyone,

I am still stuck with this issue, can anyone point me in the right
direction?

Thanks & Regards
Biplob Biswas

On Mon, Jan 18, 2016 at 2:24 PM, Biplob Biswas 
wrote:

> Hi Till,
>
> I am using flink 0.10.1 and if i am not wrong it corresponds to the
> 1.0-Snapshot you mentioned.
>
> [image: Inline image 1]
>
> If wrong, please suggest what should I do to fix it.
>
> Thanks & Regards
> Biplob Biswas
>
> On Mon, Jan 18, 2016 at 11:23 AM, Till Rohrmann 
> wrote:
>
>> Hi Biplob,
>>
>> which version of Flink are you using? With version 1.0-SNAPSHOT, I
>> cannot reproduce your problem.
>>
>> Cheers,
>> Till
>> ​
>>
>> On Sun, Jan 17, 2016 at 4:56 PM, Biplob Biswas 
>> wrote:
>>
>>> Hi,
>>>
>>> I am getting the following exception when i am using the map function
>>>
>>> Exception in thread "main"
 org.apache.flink.api.common.functions.InvalidTypesException: The return
 type of function 'computeWeightedDistribution(GraphWeighted.java:73)' could
 not be determined automatically, due to type erasure. You can give type
 information hints by using the returns(...) method on the result of the
 transformation call, or by letting your function implement the
 'ResultTypeQueryable' interface.
 at org.apache.flink.api.java.DataSet.getType(DataSet.java:176)
 at org.apache.flink.api.java.DataSet.groupBy(DataSet.java:692)
 at aim3.GraphWeighted.computeWeightedDistribution(GraphWeighted.java:74)
 at aim3.SlashdotZooInDegree.main(SlashdotZooInDegree.java:39)
 Caused by: org.apache.flink.api.common.functions.InvalidTypesException:
 Input mismatch: Basic type 'Integer' expected but was 'Long'.
 at
 org.apache.flink.api.java.typeutils.TypeExtractor.validateInputType(TypeExtractor.java:767)
 at
 org.apache.flink.api.java.typeutils.TypeExtractor.getUnaryOperatorReturnType(TypeExtractor.java:276)
 at
 org.apache.flink.api.java.typeutils.TypeExtractor.getMapReturnTypes(TypeExtractor.java:110)
 at org.apache.flink.api.java.DataSet.map(DataSet.java:213)
 at aim3.GraphWeighted.computeWeightedDistribution(GraphWeighted.java:73)
 ... 1 more
>>>
>>>
>>>
>>> This is the part of the code which I am trying to run :
>>>
>>> DataSet> distinctVertex = sourceVertex
  .union(destinationVertex)
  .groupBy(0)
  .aggregate(Aggregations.SUM, 1);
 // Compute the degrees (degree, count)

  DataSet> degreeCount = distinctVertex
  .map(new DegreeMapper())
  .groupBy(0)
  .aggregate(Aggregations.SUM, 1);
>>>
>>>
>>>
>>> and the error I am getting is at this line *.map(new DegreeMapper())*
>>>
>>> Also, the degree mapper is a simply map function which emits the second
>>> column and 1 as follows:
>>>

 public static class DegreeMapper implements
 MapFunction, Tuple2> {
 private static final long serialVersionUID = 1L;
 public Tuple2 map(Tuple2 input) throws
 Exception {
 return new Tuple2(input.f1, 1);
 }
 }
>>>
>>>
>>>
>>> Now I am lost as to what I did wrong and why I am getting that error,
>>> any help would be appreciated.
>>>
>>> Thanks a lot.
>>>
>>> Thanks & Regards
>>> Biplob Biswas
>>>
>>
>>
>

Re: integration with a scheduler

[Stephan Ewen]

If you want to programmatically start / stop / cancel jobs, have a look at
the class

"Client" (
https://github.com/apache/flink/blob/master/flink-clients/src/main/java/org/apache/flink/client/program/Client.java
)

>From the classes RemoteEnvironment or the RemoteExecutor, you can see how
to use it to send jobs and commands to a Flink cluster.




On Wed, Jan 20, 2016 at 3:03 PM, Robert Metzger  wrote:

> Hi Serkan,
>
> I would suggest to have a look at the "./bin/flink" tool. It allows you to
> start ("run") and stop ("cancel") batch and streaming jobs.
> Flink doesn't support suspending jobs.
>
> You can also use the JobManager web interface (default port: 8081) to get
> the status of the job and also to cancel it.
>
> If you need programmatic access to the job status, have a look at the
> CliFrontend class. Its the class called by the "./bin/flink" tool. There,
> you can see how jobs are submitted and cancelled.
>
> Regards,
> Robert
>
>
> 2016-01-19 12:10 GMT+01:00 :
>
>> Hi,
>>
>> I am planning to integrate flink with our job scheduler product to execute
>> jobs - especially bathc like - on flink which may be the part of some
>> other DAG style job chain.
>>
>> I need some control ablities like start, stop, suspend, get status...
>>
>> Where shold i go through ?
>>
>> --
>> Serkan Tas
>> Likya Bilgi Teknolojileri
>> ve Iletişim Hiz. Ltd.
>> www.likyateknoloji.com
>> Tel : 0 216 471 81 55
>> Gsm : 0 542 242 00 92
>> Faks:  0 216 661 14 92
>>
>> --
>> Bu elektronik posta ve onunla iletilen bütün dosyalar gizlidir. Sadece
>> yukarıda isimleri belirtilen kişiler arasında özel haberleşme amacını
>> taşımaktadır. Size yanlışlıkla ulaşmışsa bu elektonik postanın içeriğini
>> açıklamanız, kopyalamanız, yönlendirmeniz ve kullanmanız kesinlikle
>> yasaktır. Lütfen mesajı geri gönderiniz ve sisteminizden siliniz. Likya
>> Bilgi Teknolojileri ve İletişim Hiz. Ltd. Şti. bu mesajın içeriği ile
>> ilgili olarak hiç bir hukuksal sorumluluğu kabul etmez.
>>
>> This electonic mail and any files transmitted with it are intended for the
>> private use of  the persons named above. If you received this message in
>> error, forwarding, copying or use of any of the information is strictly
>> prohibited. Please immediately notify the sender and delete it from your
>> system. Likya Bilgi Teknolojileri ve İletişim Hiz. Ltd. Şti. does not
>> accept legal responsibility for the contents of this message.
>> --
>>
>> 
>> Bu e-postayı yazdırmadan önce, çevreye olan sorumluluğunuzu tekrar
>> düşünün.
>> Please consider your environmental responsibility before printing this
>> e-mail.
>>
>>
>>
>>
>

Re: Results of testing Flink quickstart against 0.10-SNAPSHOT and 1.0-SNAPSHOT (re. Dependency on non-existent org.scalamacros:quasiquotes_2.11:)

[Stephan Ewen]

Hi Prez!

I merged the pull request into master a while back. Have a look here (
https://github.com/apache/flink/commits/master commits of January 15th).

Is it possible that you are using a cached older version?

Greetings,
Stephan




On Wed, Jan 20, 2016 at 4:00 PM, Prez Cannady 
wrote:

> Morning, Robert.
>
> You’re right; the *1.0-SNAPSHOT with fetched binaries* issue is resolved
> now.  Unfortunately, it now emits the same error as *0.10-SNAPSHOT with
> fetched binaries*.  There is a fix for that:
>
> https://github.com/apache/flink/pull/1511
>
> It’s definitely in the release-0.10 and master branches
>
>
> https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorSnapshot.java
>
> https://github.com/apache/flink/blob/release-0.10/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorSnapshot.java
>
> I grabbed the sources jar for the latest 1.0-SNAPSHOT and 0.10-SNAPSHOT:
>
> From
> https://repository.apache.org/content/repositories/snapshots/org/apache/flink/flink-runtime_2.11/0.10-SNAPSHOT/flink-runtime_2.11-0.10-20160114.200924-161-sources.jar
>
> /**
>
>  * Gets the Flink (internal) accumulators values.
>
>  * @return the serialized map
>
>  */
>
> public Map>
> deserializeFlinkAccumulators() throws IOException, ClassNotFoundException {
>
> return flinkAccumulators.deserializeValue(
> *ClassLoader.getSystemClassLoader()*);
>
> }
>
> *ClassLoader.getSystemClassLoader()* should be
> *getClass().getClassLoader().*
> Not sure why it’s not taking in the build, but there’s the problem.
>
> Prez Cannady
> p: 617 500 3378
> e: revp...@opencorrelate.org
> GH: https://github.com/opencorrelate
> LI: https://www.linkedin.com/in/revprez
>
>
>
>
>
>
>
>
>
> On Jan 20, 2016, at 8:17 AM, Robert Metzger  wrote:
>
> Hi Prez,
>
> thanks a lot for the thorough research you did on this issue. The issue
> with "1.0-SNAPSHOT with fetched binary dependencies" should be resolved by
> a fix I've pushed to master yesterday:
>
> a) The "change-scala-version" script wasn't adopted to the renamed
> examples directory, that's why it failed renaming the artifacts for _2.11.
> That's why the maven dependencies ended up being mixed between Scala 2.11
> and 2.10.
>
> https://github.com/apache/flink/commit/8f0c47df092ccdc6028dbd88aed197edcd8945ee#diff-c1ca4095c51fcc58f380c59cfabffc8a
>
> b) The deployment of the scala 2.11 build failed because the "flink-table"
> module contained java code in the scala/ directory. Also the code was not
> compliant with our java checkstyle.
>
> https://github.com/apache/flink/commit/018f1fee9c856c66454155043006d6f53d39517a
>
> I hope with these two fixes, the scala 2.11 problems are resolved for now.
>
> Regards,
> Robert
>
> On Tue, Jan 19, 2016 at 6:43 AM, Prez Cannady 
> wrote:
>
>> One correction, line 3 of *1.0-SHAPSHOT source build* should read “checked
>> out *master* branch (snapshot version *1.0-SNAPSHOT*)."
>>
>> Prez Cannady
>> p: 617 500 3378
>> e: revp...@opencorrelate.org
>> GH: https://github.com/opencorrelate
>> LI: https://www.linkedin.com/in/revprez
>>
>>
>>
>>
>>
>>
>>
>>
>>
>> On Jan 19, 2016, at 12:41 AM, Prez Cannady 
>> wrote:
>>
>> Sent this to d...@flink.apache.org, but that might not be the appropriate
>> forum for it.
>>
>> Finally got a chance to sit down and run through a few tests. Upfront, I
>> have been able to resolve my issue sufficiently to move forward, but seems
>> there’s an issue with the current bits for both 1.0-SNAPSHOT and
>> 0.10-SNAPSHOT in the remote Maven repos.
>>
>> *Notes*
>>
>>1. wordcount-processing
>> is a customized
>>version of the Flink quickstart archetype I’m using to test Flink
>>integration with Spring Boot. It is instrumented for Maven and Gradle 
>> build
>>and execution.
>>2. I’m targeting Scala 2.11 and Flink 0.10.
>>
>> *0.10-SNAPSHOT source build*
>>
>> *Steps*
>>
>>1. Checked out *release–0.10* branch (snapshot version *0.10-SNAPSHOT*
>>).
>>2. Built with mvn clean install -DskipTests=true
>>-Dmaven.javadoc.skip=true -Dscala.version=2.11.7 
>> -Dscala.binary.version=2.11
>>.
>>3. Ran wordcount-process with mvn clean spring-boot:run
>>-Drun.arguments=“localhost,”.
>>4. Ran wordcount-process with gradle bootRun
>>-Drun.arguments=“localhost ”.
>>
>> *Result*
>>
>>1. Maven execution of test succeeds without incident.
>>2. Gradle execution of test succeeds without incident.
>>
>> *0.10-SNAPSHOT with fetched binary dependencies*
>>
>> *Steps*
>>
>>1. Cleaned out local maven repository with rm -rf
>>$HOME/.m2/repo/org/apache/flink && rm -rf $HOME/.m2/repo/org/scala*.
>>2. Cleaned out local gradle repository with rm -rf
>>$HOME/.gradle/caches/modules-2/files-2.1/org.apache.flink && rm -rf
>>$HOME/.gradle/caches/modules-2/files-2.1/org.scala*.
>>3. Ran wordcount-process with m

Re: Unexpected out of bounds error in UnilateralSortMerger

[Stephan Ewen]

Can you again post the stack trace? With the patched branch, the reference
mapper should not be used any more (which is where the original exception
occurred).

On Wed, Jan 20, 2016 at 7:38 PM, Theodore Vasiloudis <
theodoros.vasilou...@gmail.com> wrote:

> Alright I will try to do that.
>
> I've tried running the job with a CSV file as input, and using
> DenseVectors to represent the features, still the same IndexOutOfBounds
> error.
>
> On Wed, Jan 20, 2016 at 6:05 PM, Till Rohrmann 
> wrote:
>
>> You could change the version of Stephan’s branch via mvn versions:set
>> -DnewVersion=MyCustomBuildVersion and then mvn versions:commit. Now
>> after you install the Flink binaries you can reference them in your project
>> by setting the version of your Flink dependencies to MyCustomBuildVersion.
>> That way, you are sure that the right dependencies are used.
>>
>> Alternatively, you could compile an example program with example input
>> data which can reproduce the problem. Then I could also take a look at it.
>>
>> Cheers,
>> Till
>> ​
>>
>> On Wed, Jan 20, 2016 at 5:58 PM, Theodore Vasiloudis <
>> theodoros.vasilou...@gmail.com> wrote:
>>
>>> OK here's what I tried:
>>>
>>> * Build Flink (mvn clean install) from the branch you linked (kryo)
>>> * Build my uber-jar, I use SBT with 1.0-SNAPSHOT as the Flink version,
>>> added local maven repo to resolvers so that it picks up the previously
>>> installed version (I hope)
>>> * Launch local cluster from newly built Flink, try to run job
>>>
>>> Still getting the same error.
>>>
>>> Is there a way to ensure that SBT is picking up the local version of
>>> Flink to build the uber-jar?
>>> Does it matter in this case, or is it enough that I'm sure the launched
>>> Flink instance comes from the branch you linked?
>>>
>>>
>>> On Wed, Jan 20, 2016 at 4:30 PM, Stephan Ewen  wrote:
>>>
 The bug looks to be in the serialization via Kryo while spilling
 windows. Note that Kryo is here used as a fallback serializer, since the
 SparseVector is not transparent type to Flink.

 I think there are two possible reasons:
   1) Kryo, or our Kryo setup has an issue here
   2) Kryo is inconsistently configured. There are multiple Kryo
 instances used across the serializers in the sorter. There may be a bug
 that they are not initialized in sync.


 To check this, can you build Flink with this pull request (
 https://github.com/apache/flink/pull/1528) or from this branch (
 https://github.com/StephanEwen/incubator-flink kryo) and see if that
 fixes it?


 Thanks,
 Stephan





 On Wed, Jan 20, 2016 at 3:33 PM, Theodore Vasiloudis <
 theodoros.vasilou...@gmail.com> wrote:

> I haven't been able to reproduce this with other datasets. Taking a
> smaller sample from the large dataset I'm using (link to data
> )
> causes the same problem however.
>
> I'm wondering if the implementation of readLibSVM is what's wrong
> here. I've tried the new version commited recently by Chiwan, but I still
> get the same error.
>
> I'll see if I can spot a bug in readLibSVM.
>
> On Wed, Jan 20, 2016 at 1:43 PM, Theodore Vasiloudis <
> theodoros.vasilou...@gmail.com> wrote:
>
>> It's on 0.10.
>>
>> I've tried explicitly registering SparseVector (which is done anyway
>> by registerFlinkMLTypes
>> 
>> which is called when the SVM predict or evaluate functions are called
>> )
>> in my job but I still get the same. I will try a couple different 
>> datasets
>> and try to see if it's the number of features that is causing this or
>> something else.
>>
>> So far it works fine for a dataset with 8 features, but the large one
>> has 2000 and I get the above error there. I will try large datasets with 
>> a
>> few features and small datasets with many features as well.
>>
>> On Wed, Jan 20, 2016 at 11:39 AM, Stephan Ewen 
>> wrote:
>>
>>> Hi!
>>>
>>> Does this error occur in 0.10 or im 1.0-SNAPSHOT?
>>>
>>> It is probably an incorrectly configured Kryo instance (not a
>>> problem of the sorter).
>>> What is strange is that it occurs in the "MapReferenceResolver" -
>>> there should be no reference resolution during serialization /
>>> deserialization.
>>>
>>> Can you try what happens when you explicitly register the type
>>> SparseVector at the ExecutionEnvironment?
>>>
>>> Stephan
>>>
>>>
>>> On Wed, Jan 

Re: Error starting job manager in 1.0-SNAPSHOT

[Stephan Ewen]

Sorry for the confusion with that. The 1.0-SNAPSHOT is changing quite a
bit, we are trying to consolidate as much as possible for 1.0 to keep
braking changes after that low.

On Wed, Jan 20, 2016 at 7:58 PM, Andrew Whitaker <
andrew.whita...@braintreepayments.com> wrote:

> Stephen,
>
> Thanks so much for the quick response. That worked for me!
>
> On Wed, Jan 20, 2016 at 11:34 AM, Stephan Ewen  wrote:
>
>> Hi!
>>
>> As of a few weeks ago, there is no "streaming" or "batch" mode any more.
>> There is only one mode that handles both.
>>
>> I think the argument "streaming" passed to the script is then incorrectly
>> interpreted as the hostname to bin the JobManager network interface to.
>> Then you get the "UnknownHostException".
>>
>> Simply drop "streaming" from the command line arguments (call 
>> ./bin/jobmanager.sh
>> start cluster). That should solve it.
>>
>> Best,
>> Stephan
>>
>>
>> On Wed, Jan 20, 2016 at 6:23 PM, Andrew Whitaker <
>> andrew.whita...@braintreepayments.com> wrote:
>>
>>> Hi,
>>>
>>> I'm getting the following error when attempting to start the job manager:
>>>
>>> ```
>>> ./bin/jobmanager.sh start cluster streaming
>>> ```
>>>
>>> ```
>>> 10:51:27,824 INFO  org.apache.flink.runtime.jobmanager.JobManager
>>>  - Registered UNIX signal handlers for [TERM, HUP, INT]
>>> 10:51:27,914 INFO  org.apache.flink.runtime.jobmanager.JobManager
>>>  - Loading configuration from
>>> /Users/anwhitaker/Downloads/flink-1.0-SNAPSHOT 3/conf
>>> 10:51:27,922 INFO  org.apache.flink.runtime.jobmanager.JobManager
>>>  - Starting JobManager without high-availability
>>> 10:51:28,034 ERROR org.apache.flink.runtime.jobmanager.JobManager
>>>  - streaming: unknown error
>>> java.net.UnknownHostException: streaming: unknown error
>>> at java.net.Inet6AddressImpl.lookupAllHostAddr(Native Method)
>>> at java.net.InetAddress$2.lookupAllHostAddr(InetAddress.java:928)
>>> at
>>> java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1323)
>>> at java.net.InetAddress.getAllByName0(InetAddress.java:1276)
>>> at java.net.InetAddress.getAllByName(InetAddress.java:1192)
>>> at java.net.InetAddress.getAllByName(InetAddress.java:1126)
>>> at java.net.InetAddress.getByName(InetAddress.java:1076)
>>> at
>>> org.apache.flink.runtime.jobmanager.JobManager$.parseArgs(JobManager.scala:1955)
>>> at
>>> org.apache.flink.runtime.jobmanager.JobManager$.liftedTree2$1(JobManager.scala:1517)
>>> at
>>> org.apache.flink.runtime.jobmanager.JobManager$.main(JobManager.scala:1516)
>>> at org.apache.flink.runtime.jobmanager.JobManager.main(JobManager.scala)
>>> ```
>>>
>>> I don't think my config has changed. Are there changes from the last few
>>> days that could be causing this?
>>>
>>> Thanks,
>>>
>>> Andrew Whitaker | andrew.whita...@braintreepayments.com
>>>
>>
>>
>
>
> --
> Andrew Whitaker | andrew.whita...@braintreepayments.com
> --
> Note: this information is confidential. It is prohibited to share, post
> online or otherwise publicize without Braintree's prior written consent.
>

Re: DeserializationSchema isEndOfStream usage?

[Robert Metzger]

I've now merged the pull request. DeserializationSchema.isEndOfStream()
should now be evaluated correctly by the Kafka 0.9 and 0.8 connectors.

Please let me know if the updated code has any issues. I'll fix the issues
asap.

On Wed, Jan 13, 2016 at 5:06 PM, David Kim 
wrote:

> Thanks Robert! I'll be keeping tabs on the PR.
>
> Cheers,
> David
>
> On Mon, Jan 11, 2016 at 4:04 PM, Robert Metzger 
> wrote:
>
>> Hi David,
>>
>> In theory isEndOfStream() is absolutely the right way to go for stopping
>> data sources in Flink.
>> That its not working as expected is a bug. I have a pending pull request
>> for adding a Kafka 0.9 connector, which fixes this issue as well (for all
>> supported Kafka versions).
>>
>> Sorry for the inconvenience. If you want, you can check out the branch of
>> the PR and build Flink yourself to get the fix.
>> I hope that I can merge the connector to master this week, then, the fix
>> will be available in 1.0-SNAPSHOT as well.
>>
>> Regards,
>> Robert
>>
>>
>>
>> Sent from my iPhone
>>
>> On 11.01.2016, at 21:39, David Kim 
>> wrote:
>>
>> Hello all,
>>
>> I saw that DeserializationSchema has an API "isEndOfStream()".
>>
>>
>> https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/DeserializationSchema.java
>>
>> Can *isEndOfStream* be utilized to somehow terminate a streaming flink
>> job?
>>
>> I was under the impression that if we return "true" we can control when a
>> stream can close. The use case I had in mind was controlling when
>> unit/integration tests would terminate a flink job. We can rely on the fact
>> that a test/spec would know how many items it expects to consume and then
>> switch *isEndOfStream* to return true.
>>
>> Am I misunderstanding the intention for *isEndOfStream*?
>>
>> I also set a breakpoint on *isEndOfStream* and saw that it never was hit
>> when using "FlinkKafkaConsumer082" to pass in a DeserializationSchema
>> implementation.
>>
>> Currently testing on 1.0-SNAPSHOT.
>>
>> Cheers!
>> David
>>
>>
>
>
> --
> Note: this information is confidential. It is prohibited to share, post
> online or otherwise publicize without Braintree's prior written consent.
>

Re: Error starting job manager in 1.0-SNAPSHOT

[Andrew Whitaker]

Stephen,

Thanks so much for the quick response. That worked for me!

On Wed, Jan 20, 2016 at 11:34 AM, Stephan Ewen  wrote:

> Hi!
>
> As of a few weeks ago, there is no "streaming" or "batch" mode any more.
> There is only one mode that handles both.
>
> I think the argument "streaming" passed to the script is then incorrectly
> interpreted as the hostname to bin the JobManager network interface to.
> Then you get the "UnknownHostException".
>
> Simply drop "streaming" from the command line arguments (call 
> ./bin/jobmanager.sh
> start cluster). That should solve it.
>
> Best,
> Stephan
>
>
> On Wed, Jan 20, 2016 at 6:23 PM, Andrew Whitaker <
> andrew.whita...@braintreepayments.com> wrote:
>
>> Hi,
>>
>> I'm getting the following error when attempting to start the job manager:
>>
>> ```
>> ./bin/jobmanager.sh start cluster streaming
>> ```
>>
>> ```
>> 10:51:27,824 INFO  org.apache.flink.runtime.jobmanager.JobManager
>>- Registered UNIX signal handlers for [TERM, HUP, INT]
>> 10:51:27,914 INFO  org.apache.flink.runtime.jobmanager.JobManager
>>- Loading configuration from
>> /Users/anwhitaker/Downloads/flink-1.0-SNAPSHOT 3/conf
>> 10:51:27,922 INFO  org.apache.flink.runtime.jobmanager.JobManager
>>- Starting JobManager without high-availability
>> 10:51:28,034 ERROR org.apache.flink.runtime.jobmanager.JobManager
>>- streaming: unknown error
>> java.net.UnknownHostException: streaming: unknown error
>> at java.net.Inet6AddressImpl.lookupAllHostAddr(Native Method)
>> at java.net.InetAddress$2.lookupAllHostAddr(InetAddress.java:928)
>> at java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1323)
>> at java.net.InetAddress.getAllByName0(InetAddress.java:1276)
>> at java.net.InetAddress.getAllByName(InetAddress.java:1192)
>> at java.net.InetAddress.getAllByName(InetAddress.java:1126)
>> at java.net.InetAddress.getByName(InetAddress.java:1076)
>> at
>> org.apache.flink.runtime.jobmanager.JobManager$.parseArgs(JobManager.scala:1955)
>> at
>> org.apache.flink.runtime.jobmanager.JobManager$.liftedTree2$1(JobManager.scala:1517)
>> at
>> org.apache.flink.runtime.jobmanager.JobManager$.main(JobManager.scala:1516)
>> at org.apache.flink.runtime.jobmanager.JobManager.main(JobManager.scala)
>> ```
>>
>> I don't think my config has changed. Are there changes from the last few
>> days that could be causing this?
>>
>> Thanks,
>>
>> Andrew Whitaker | andrew.whita...@braintreepayments.com
>>
>
>


-- 
Andrew Whitaker | andrew.whita...@braintreepayments.com
--
Note: this information is confidential. It is prohibited to share, post
online or otherwise publicize without Braintree's prior written consent.

Re: Unexpected out of bounds error in UnilateralSortMerger

[Theodore Vasiloudis]

Alright I will try to do that.

I've tried running the job with a CSV file as input, and using DenseVectors
to represent the features, still the same IndexOutOfBounds error.

On Wed, Jan 20, 2016 at 6:05 PM, Till Rohrmann  wrote:

> You could change the version of Stephan’s branch via mvn versions:set
> -DnewVersion=MyCustomBuildVersion and then mvn versions:commit. Now after
> you install the Flink binaries you can reference them in your project by
> setting the version of your Flink dependencies to MyCustomBuildVersion.
> That way, you are sure that the right dependencies are used.
>
> Alternatively, you could compile an example program with example input
> data which can reproduce the problem. Then I could also take a look at it.
>
> Cheers,
> Till
> ​
>
> On Wed, Jan 20, 2016 at 5:58 PM, Theodore Vasiloudis <
> theodoros.vasilou...@gmail.com> wrote:
>
>> OK here's what I tried:
>>
>> * Build Flink (mvn clean install) from the branch you linked (kryo)
>> * Build my uber-jar, I use SBT with 1.0-SNAPSHOT as the Flink version,
>> added local maven repo to resolvers so that it picks up the previously
>> installed version (I hope)
>> * Launch local cluster from newly built Flink, try to run job
>>
>> Still getting the same error.
>>
>> Is there a way to ensure that SBT is picking up the local version of
>> Flink to build the uber-jar?
>> Does it matter in this case, or is it enough that I'm sure the launched
>> Flink instance comes from the branch you linked?
>>
>>
>> On Wed, Jan 20, 2016 at 4:30 PM, Stephan Ewen  wrote:
>>
>>> The bug looks to be in the serialization via Kryo while spilling
>>> windows. Note that Kryo is here used as a fallback serializer, since the
>>> SparseVector is not transparent type to Flink.
>>>
>>> I think there are two possible reasons:
>>>   1) Kryo, or our Kryo setup has an issue here
>>>   2) Kryo is inconsistently configured. There are multiple Kryo
>>> instances used across the serializers in the sorter. There may be a bug
>>> that they are not initialized in sync.
>>>
>>>
>>> To check this, can you build Flink with this pull request (
>>> https://github.com/apache/flink/pull/1528) or from this branch (
>>> https://github.com/StephanEwen/incubator-flink kryo) and see if that
>>> fixes it?
>>>
>>>
>>> Thanks,
>>> Stephan
>>>
>>>
>>>
>>>
>>>
>>> On Wed, Jan 20, 2016 at 3:33 PM, Theodore Vasiloudis <
>>> theodoros.vasilou...@gmail.com> wrote:
>>>
 I haven't been able to reproduce this with other datasets. Taking a
 smaller sample from the large dataset I'm using (link to data
 )
 causes the same problem however.

 I'm wondering if the implementation of readLibSVM is what's wrong here.
 I've tried the new version commited recently by Chiwan, but I still get the
 same error.

 I'll see if I can spot a bug in readLibSVM.

 On Wed, Jan 20, 2016 at 1:43 PM, Theodore Vasiloudis <
 theodoros.vasilou...@gmail.com> wrote:

> It's on 0.10.
>
> I've tried explicitly registering SparseVector (which is done anyway
> by registerFlinkMLTypes
> 
> which is called when the SVM predict or evaluate functions are called
> )
> in my job but I still get the same. I will try a couple different datasets
> and try to see if it's the number of features that is causing this or
> something else.
>
> So far it works fine for a dataset with 8 features, but the large one
> has 2000 and I get the above error there. I will try large datasets with a
> few features and small datasets with many features as well.
>
> On Wed, Jan 20, 2016 at 11:39 AM, Stephan Ewen 
> wrote:
>
>> Hi!
>>
>> Does this error occur in 0.10 or im 1.0-SNAPSHOT?
>>
>> It is probably an incorrectly configured Kryo instance (not a problem
>> of the sorter).
>> What is strange is that it occurs in the "MapReferenceResolver" -
>> there should be no reference resolution during serialization /
>> deserialization.
>>
>> Can you try what happens when you explicitly register the type
>> SparseVector at the ExecutionEnvironment?
>>
>> Stephan
>>
>>
>> On Wed, Jan 20, 2016 at 11:24 AM, Theodore Vasiloudis <
>> theodoros.vasilou...@gmail.com> wrote:
>>
>>> Hello all,
>>>
>>> I'm trying to run a job using FlinkML and I'm confused about the
>>> source of an error.
>>>
>>> The job reads a libSVM formatted file and trains an SVM classifier
>>> on it.
>>>
>>> I've tried this with small datasets and everything works o

Re: An interesting apache project: Reef

[Stephan Ewen]

Thanks for the pointers.

I actually know some of the Reef people (like Markus Weimer) and we talked
about ideas a few times. Especially around batch and iterations. There were
many common thoughts and good discussions.

In the streaming space, I think Flink and Reef diverged quite a bit
(checkpointing, latency, state).

On Wed, Jan 20, 2016 at 5:21 PM, kovas boguta 
wrote:

> Some people here (especially Flink contributors) might be interested to
> know about this project: https://reef.apache.org/index.html
>
> It is lower-level than Flink (and less mature), but with similar
> architectural sensibilities and emphasis on interfaces. It would be pretty
> interesting to compare the designs in areas of overlap, and maybe some
> cross-pollination of ideas (if not outright reuse/compatibility) is
> possible.
>
> Also check out the related subprojects:
> https://reef.apache.org/tang.html
> https://reef.apache.org/wake.html
>
>
>
>

Re: JDBCInputFormat GC overhead limit exceeded error

[Stephan Ewen]

Super, thanks for finding this. Makes a lot of sense to have a result set
that does hold onto data.

Would be great if you could open a pull request with this fix, as other
users will benefit from that as well!

Cheers,
Stephan


On Wed, Jan 20, 2016 at 6:03 PM, Maximilian Bode <
maximilian.b...@tngtech.com> wrote:

> Hi Stephan,
>
> thanks for your fast answer. Just setting the Flink-managed memory to a
> low value would not have worked for us, as we need joins etc. in the same
> job.
>
> After investigating the JDBCInputFormat, we found the line
> statement = dbConn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE,
> ResultSet.CONCUR_READ_ONLY);
> to be the culprit; to be more exact, the scrollable result set. When
> replaced with TYPE_FORWARD_ONLY, some changes have to be made to
> nextRecord() and reachedEnd(), but this does the job without memory leak.
>
> Another change that might be useful (as far as performance is concerned)
> is disabling autocommits and letting users decide the fetchSize (somewhat
> in parallel to batchInterval in JDBCOutputFormat).
>
> Cheers,
> Max
> —
> Maximilian Bode * Junior Consultant * maximilian.b...@tngtech.com * 0176
> 1000 75 50
> TNG Technology Consulting GmbH, Betastr. 13a, 85774 Unterföhring
> Geschäftsführer: Henrik Klagges, Christoph Stock, Dr. Robert Dahlke
> Sitz: Unterföhring * Amtsgericht München * HRB 135082
>
> Am 19.01.2016 um 21:26 schrieb Stephan Ewen :
>
> Hi!
>
> This kind of error (GC overhead exceeded) usually means that the system is
> reaching a state where it has very many still living objects and frees
> little memory during each collection. As a consequence, it is basically
> busy with only garbage collection.
>
> Your job probably has about 500-600 MB or free memory, the rest is at that
> memory size reserved for JVM overhead and Flink's worker memory.
> Now, since your job actually does not keep any objects or rows around,
> this should be plenty. I can only suspect that the Oracle JDBC driver is
> very memory hungry, thus pushing the system to the limit. (I found this,
> for example
>
> What you can do:
>  For this kind of job, you can simply tell Flink to not reserve as much
> memory, by using the option "taskmanager.memory.size=1". If the JDBC driver
> has no leak, but is simply super hungry, this should solve it.
>
> Greetings,
> Stephan
>
>
> I also found these resources concerning Oracle JDBC memory:
>
>  -
> http://stackoverflow.com/questions/2876895/oracle-t4cpreparedstatement-memory-leaks
> (bottom answer)
>  - https://community.oracle.com/thread/2220078?tstart=0
>
>
> On Tue, Jan 19, 2016 at 5:44 PM, Maximilian Bode <
> maximilian.b...@tngtech.com> wrote:
>
>> Hi Robert,
>>
>> I am using 0.10.1.
>>
>>
>> Am 19.01.2016 um 17:42 schrieb Robert Metzger :
>>
>> Hi Max,
>>
>> which version of Flink are you using?
>>
>> On Tue, Jan 19, 2016 at 5:35 PM, Maximilian Bode <
>> maximilian.b...@tngtech.com> wrote:
>>
>>> Hi everyone,
>>>
>>> I am facing a problem using the JDBCInputFormat which occurred in a
>>> larger Flink job. As a minimal example I can reproduce it when just writing
>>> data into a csv after having read it from a database, i.e.
>>>
>>> DataSet> existingData = env.createInput(
>>> JDBCInputFormat.buildJDBCInputFormat()
>>> .setDrivername("oracle.jdbc.driver.OracleDriver")
>>> .setUsername(…)
>>> .setPassword(…)
>>> .setDBUrl(…)
>>> .setQuery("select DATA from TABLENAME")
>>> .finish(),
>>> new TupleTypeInfo<>(Tuple1.class, BasicTypeInfo.STRING_TYPE_INFO));
>>> existingData.writeAsCsv(…);
>>>
>>> where DATA is a column containing strings of length ~25 characters and
>>> TABLENAME contains 20 million rows.
>>>
>>> After starting the job on a YARN cluster (using -tm 3072 and leaving the
>>> other memory settings at default values), Flink happily goes along at first
>>> but then fails after something like three million records have been sent by
>>> the JDBCInputFormat. The Exception reads "The slot in which the task was
>>> executed has been released. Probably loss of TaskManager …". The local
>>> taskmanager.log in the affected container reads
>>> "java.lang.OutOfMemoryError: GC overhead limit exceeded
>>> at
>>> java.util.Collections$UnmodifiableCollection.iterator(Collections.java:1063)
>>>
>>> at 
>>> org.jboss.netty.channel.socket.nio.NioClientBoss.processConnectTimeout(NioClientBoss.java:119)
>>> at
>>> org.jboss.netty.channel.socket.nio.NioClientBoss.process(NioClientBoss.java:83)
>>> at
>>> org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:312)
>>> at
>>> org.jboss.netty.channel.socket.nio.NioClientBoss.run(NioClientBoss.java:42)
>>> at
>>> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
>>> at
>>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
>>> at java.lang.Thread.run(Thread.java:744)"
>>>
>>> Any ideas what is going wrong here?
>>>
>>> Cheers,
>>> Max
>>>
>>> —
>>

Re: Cannot start FlinkMiniCluster.WebServer using different port in FlinkMiniCluster

[Stephan Ewen]

I think this null pointer comes when the log files are not found (bug in
0.10).

You can double check by either trying 1.0-SNAPSHOT or putting for test an
absolute path of a file that exists for the log file.

Greetings,
Stephan

On Wed, Jan 20, 2016 at 6:33 PM, Till Rohrmann  wrote:

> I guess it’s easiest to simply enable logging and see what the problem is.
> If you run it from the IDE then you can also set a breakpoint in
> WebMonitorUtils.startWebRuntimeMonitor and see what the exception is.
>
> Cheers,
> Till
> ​
>
> On Wed, Jan 20, 2016 at 6:04 PM, HungChang 
> wrote:
>
>> Yea I'm wondering why the web server cannot be instantiated because
>> changing
>> the port 8081 to  works well in the following demo sample of Flink.
>>
>> https://github.com/dataArtisans/flink-streaming-demo/blob/master/src/main/scala/com/dataartisans/flink_demo/utils/DemoStreamEnvironment.scala
>>
>> so is the code of Flink syntax correct and I should try to find other
>> issues?
>>
>> Best,
>>
>> Sendoh
>>
>>
>>
>> --
>> View this message in context:
>> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Cannot-start-FlinkMiniCluster-WebServer-using-different-port-in-FlinkMiniCluster-tp4414p4420.html
>> Sent from the Apache Flink User Mailing List archive. mailing list
>> archive at Nabble.com.
>>
>
>

Re: Reading Binary Data (Matrix) with Flink

[Till Rohrmann]

With readHadoopFile you can use all of Hadoop’s FileInputFormats and thus
you can also do everything with Flink, what you can do with Hadoop. Simply
take the same Hadoop FileInputFormat which you would take for your
MapReduce job.

Cheers,
Till
​

On Wed, Jan 20, 2016 at 3:16 PM, Saliya Ekanayake  wrote:

> Thank you, I saw the readHadoopFile, but I was not sure how it can be used
> to the following, which is what I need. The logic of the code requires an
> entire row to operate on, so in our current implementation with P tasks,
> each of them will read a rectangular block of (N/P) x N from the matrix. Is
> this possible with readHadoopFile? Also, the file may not be in hdfs, so is
> it possible to refer to local disk in doing this?
>
> Thank you
>
> On Wed, Jan 20, 2016 at 1:31 AM, Chiwan Park 
> wrote:
>
>> Hi Saliya,
>>
>> You can use the input format from Hadoop in Flink by using readHadoopFile
>> method. The method returns a dataset which of type is Tuple2.
>> Note that MapReduce equivalent transformation in Flink is composed of map,
>> groupBy, and reduceGroup.
>>
>> > On Jan 20, 2016, at 3:04 PM, Suneel Marthi  wrote:
>> >
>> > Guess u r looking for Flink's BinaryInputFormat to be able to read
>> blocks of data from HDFS
>> >
>> >
>> https://ci.apache.org/projects/flink/flink-docs-release-0.10/api/java/org/apache/flink/api/common/io/BinaryInputFormat.html
>> >
>> > On Wed, Jan 20, 2016 at 12:45 AM, Saliya Ekanayake 
>> wrote:
>> > Hi,
>> >
>> > I am trying to use Flink perform a parallel batch operation on a NxN
>> matrix represented as a binary file. Each (i,j) element is stored as a Java
>> Short value. In a typical MapReduce programming with Hadoop, each map task
>> will read a block of rows of this matrix and perform computation on that
>> block and emit result to the reducer.
>> >
>> > How is this done in Flink? I am new to Flink and couldn't find a binary
>> reader so far. Any help is greatly appreciated.
>> >
>> > Thank you,
>> > Saliya
>> >
>> > --
>> > Saliya Ekanayake
>> > Ph.D. Candidate | Research Assistant
>> > School of Informatics and Computing | Digital Science Center
>> > Indiana University, Bloomington
>> > Cell 812-391-4914
>> > http://saliya.org
>> >
>>
>> Regards,
>> Chiwan Park
>>
>>
>
>
> --
> Saliya Ekanayake
> Ph.D. Candidate | Research Assistant
> School of Informatics and Computing | Digital Science Center
> Indiana University, Bloomington
> Cell 812-391-4914
> http://saliya.org
>

Re: Error starting job manager in 1.0-SNAPSHOT

[Stephan Ewen]

Hi!

As of a few weeks ago, there is no "streaming" or "batch" mode any more.
There is only one mode that handles both.

I think the argument "streaming" passed to the script is then incorrectly
interpreted as the hostname to bin the JobManager network interface to.
Then you get the "UnknownHostException".

Simply drop "streaming" from the command line arguments (call
./bin/jobmanager.sh
start cluster). That should solve it.

Best,
Stephan


On Wed, Jan 20, 2016 at 6:23 PM, Andrew Whitaker <
andrew.whita...@braintreepayments.com> wrote:

> Hi,
>
> I'm getting the following error when attempting to start the job manager:
>
> ```
> ./bin/jobmanager.sh start cluster streaming
> ```
>
> ```
> 10:51:27,824 INFO  org.apache.flink.runtime.jobmanager.JobManager
>- Registered UNIX signal handlers for [TERM, HUP, INT]
> 10:51:27,914 INFO  org.apache.flink.runtime.jobmanager.JobManager
>- Loading configuration from
> /Users/anwhitaker/Downloads/flink-1.0-SNAPSHOT 3/conf
> 10:51:27,922 INFO  org.apache.flink.runtime.jobmanager.JobManager
>- Starting JobManager without high-availability
> 10:51:28,034 ERROR org.apache.flink.runtime.jobmanager.JobManager
>- streaming: unknown error
> java.net.UnknownHostException: streaming: unknown error
> at java.net.Inet6AddressImpl.lookupAllHostAddr(Native Method)
> at java.net.InetAddress$2.lookupAllHostAddr(InetAddress.java:928)
> at java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1323)
> at java.net.InetAddress.getAllByName0(InetAddress.java:1276)
> at java.net.InetAddress.getAllByName(InetAddress.java:1192)
> at java.net.InetAddress.getAllByName(InetAddress.java:1126)
> at java.net.InetAddress.getByName(InetAddress.java:1076)
> at
> org.apache.flink.runtime.jobmanager.JobManager$.parseArgs(JobManager.scala:1955)
> at
> org.apache.flink.runtime.jobmanager.JobManager$.liftedTree2$1(JobManager.scala:1517)
> at
> org.apache.flink.runtime.jobmanager.JobManager$.main(JobManager.scala:1516)
> at org.apache.flink.runtime.jobmanager.JobManager.main(JobManager.scala)
> ```
>
> I don't think my config has changed. Are there changes from the last few
> days that could be causing this?
>
> Thanks,
>
> Andrew Whitaker | andrew.whita...@braintreepayments.com
>

Re: Cannot start FlinkMiniCluster.WebServer using different port in FlinkMiniCluster

[Till Rohrmann]

I guess it’s easiest to simply enable logging and see what the problem is.
If you run it from the IDE then you can also set a breakpoint in
WebMonitorUtils.startWebRuntimeMonitor and see what the exception is.

Cheers,
Till
​

On Wed, Jan 20, 2016 at 6:04 PM, HungChang 
wrote:

> Yea I'm wondering why the web server cannot be instantiated because
> changing
> the port 8081 to  works well in the following demo sample of Flink.
>
> https://github.com/dataArtisans/flink-streaming-demo/blob/master/src/main/scala/com/dataartisans/flink_demo/utils/DemoStreamEnvironment.scala
>
> so is the code of Flink syntax correct and I should try to find other
> issues?
>
> Best,
>
> Sendoh
>
>
>
> --
> View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Cannot-start-FlinkMiniCluster-WebServer-using-different-port-in-FlinkMiniCluster-tp4414p4420.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>

Error starting job manager in 1.0-SNAPSHOT

[Andrew Whitaker]

Hi,

I'm getting the following error when attempting to start the job manager:

```
./bin/jobmanager.sh start cluster streaming
```

```
10:51:27,824 INFO  org.apache.flink.runtime.jobmanager.JobManager
 - Registered UNIX signal handlers for [TERM, HUP, INT]
10:51:27,914 INFO  org.apache.flink.runtime.jobmanager.JobManager
 - Loading configuration from
/Users/anwhitaker/Downloads/flink-1.0-SNAPSHOT 3/conf
10:51:27,922 INFO  org.apache.flink.runtime.jobmanager.JobManager
 - Starting JobManager without high-availability
10:51:28,034 ERROR org.apache.flink.runtime.jobmanager.JobManager
 - streaming: unknown error
java.net.UnknownHostException: streaming: unknown error
at java.net.Inet6AddressImpl.lookupAllHostAddr(Native Method)
at java.net.InetAddress$2.lookupAllHostAddr(InetAddress.java:928)
at java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1323)
at java.net.InetAddress.getAllByName0(InetAddress.java:1276)
at java.net.InetAddress.getAllByName(InetAddress.java:1192)
at java.net.InetAddress.getAllByName(InetAddress.java:1126)
at java.net.InetAddress.getByName(InetAddress.java:1076)
at
org.apache.flink.runtime.jobmanager.JobManager$.parseArgs(JobManager.scala:1955)
at
org.apache.flink.runtime.jobmanager.JobManager$.liftedTree2$1(JobManager.scala:1517)
at
org.apache.flink.runtime.jobmanager.JobManager$.main(JobManager.scala:1516)
at org.apache.flink.runtime.jobmanager.JobManager.main(JobManager.scala)
```

I don't think my config has changed. Are there changes from the last few
days that could be causing this?

Thanks,

Andrew Whitaker | andrew.whita...@braintreepayments.com

Re: Cannot start FlinkMiniCluster.WebServer using different port in FlinkMiniCluster

[HungChang]

Yea I'm wondering why the web server cannot be instantiated because changing
the port 8081 to  works well in the following demo sample of Flink.
https://github.com/dataArtisans/flink-streaming-demo/blob/master/src/main/scala/com/dataartisans/flink_demo/utils/DemoStreamEnvironment.scala

so is the code of Flink syntax correct and I should try to find other
issues?

Best,

Sendoh



--
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Cannot-start-FlinkMiniCluster-WebServer-using-different-port-in-FlinkMiniCluster-tp4414p4420.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.

Re: Unexpected out of bounds error in UnilateralSortMerger

[Till Rohrmann]

You could change the version of Stephan’s branch via mvn versions:set
-DnewVersion=MyCustomBuildVersion and then mvn versions:commit. Now after
you install the Flink binaries you can reference them in your project by
setting the version of your Flink dependencies to MyCustomBuildVersion.
That way, you are sure that the right dependencies are used.

Alternatively, you could compile an example program with example input data
which can reproduce the problem. Then I could also take a look at it.

Cheers,
Till
​

On Wed, Jan 20, 2016 at 5:58 PM, Theodore Vasiloudis <
theodoros.vasilou...@gmail.com> wrote:

> OK here's what I tried:
>
> * Build Flink (mvn clean install) from the branch you linked (kryo)
> * Build my uber-jar, I use SBT with 1.0-SNAPSHOT as the Flink version,
> added local maven repo to resolvers so that it picks up the previously
> installed version (I hope)
> * Launch local cluster from newly built Flink, try to run job
>
> Still getting the same error.
>
> Is there a way to ensure that SBT is picking up the local version of Flink
> to build the uber-jar?
> Does it matter in this case, or is it enough that I'm sure the launched
> Flink instance comes from the branch you linked?
>
>
> On Wed, Jan 20, 2016 at 4:30 PM, Stephan Ewen  wrote:
>
>> The bug looks to be in the serialization via Kryo while spilling windows.
>> Note that Kryo is here used as a fallback serializer, since the
>> SparseVector is not transparent type to Flink.
>>
>> I think there are two possible reasons:
>>   1) Kryo, or our Kryo setup has an issue here
>>   2) Kryo is inconsistently configured. There are multiple Kryo instances
>> used across the serializers in the sorter. There may be a bug that they are
>> not initialized in sync.
>>
>>
>> To check this, can you build Flink with this pull request (
>> https://github.com/apache/flink/pull/1528) or from this branch (
>> https://github.com/StephanEwen/incubator-flink kryo) and see if that
>> fixes it?
>>
>>
>> Thanks,
>> Stephan
>>
>>
>>
>>
>>
>> On Wed, Jan 20, 2016 at 3:33 PM, Theodore Vasiloudis <
>> theodoros.vasilou...@gmail.com> wrote:
>>
>>> I haven't been able to reproduce this with other datasets. Taking a
>>> smaller sample from the large dataset I'm using (link to data
>>> )
>>> causes the same problem however.
>>>
>>> I'm wondering if the implementation of readLibSVM is what's wrong here.
>>> I've tried the new version commited recently by Chiwan, but I still get the
>>> same error.
>>>
>>> I'll see if I can spot a bug in readLibSVM.
>>>
>>> On Wed, Jan 20, 2016 at 1:43 PM, Theodore Vasiloudis <
>>> theodoros.vasilou...@gmail.com> wrote:
>>>
 It's on 0.10.

 I've tried explicitly registering SparseVector (which is done anyway by
 registerFlinkMLTypes
 
 which is called when the SVM predict or evaluate functions are called
 )
 in my job but I still get the same. I will try a couple different datasets
 and try to see if it's the number of features that is causing this or
 something else.

 So far it works fine for a dataset with 8 features, but the large one
 has 2000 and I get the above error there. I will try large datasets with a
 few features and small datasets with many features as well.

 On Wed, Jan 20, 2016 at 11:39 AM, Stephan Ewen 
 wrote:

> Hi!
>
> Does this error occur in 0.10 or im 1.0-SNAPSHOT?
>
> It is probably an incorrectly configured Kryo instance (not a problem
> of the sorter).
> What is strange is that it occurs in the "MapReferenceResolver" -
> there should be no reference resolution during serialization /
> deserialization.
>
> Can you try what happens when you explicitly register the type
> SparseVector at the ExecutionEnvironment?
>
> Stephan
>
>
> On Wed, Jan 20, 2016 at 11:24 AM, Theodore Vasiloudis <
> theodoros.vasilou...@gmail.com> wrote:
>
>> Hello all,
>>
>> I'm trying to run a job using FlinkML and I'm confused about the
>> source of an error.
>>
>> The job reads a libSVM formatted file and trains an SVM classifier on
>> it.
>>
>> I've tried this with small datasets and everything works out fine.
>>
>> When trying to run the same job on a large dataset (~11GB
>> uncompressed) however, I get the following error:
>>
>>
>>> java.lang.RuntimeException: Error obtaining the sorted input: Thread
>>> 'SortMerger spilling thread' terminated due to an exception:
>>> java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
>>

Re: JDBCInputFormat GC overhead limit exceeded error

[Maximilian Bode]

Hi Stephan,

thanks for your fast answer. Just setting the Flink-managed memory to a low 
value would not have worked for us, as we need joins etc. in the same job.

After investigating the JDBCInputFormat, we found the line 
statement = dbConn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, 
ResultSet.CONCUR_READ_ONLY);
to be the culprit; to be more exact, the scrollable result set. When replaced 
with TYPE_FORWARD_ONLY, some changes have to be made to nextRecord() and 
reachedEnd(), but this does the job without memory leak.

Another change that might be useful (as far as performance is concerned) is 
disabling autocommits and letting users decide the fetchSize (somewhat in 
parallel to batchInterval in JDBCOutputFormat).

Cheers,
Max
— 
Maximilian Bode * Junior Consultant * maximilian.b...@tngtech.com * 0176 1000 
75 50
TNG Technology Consulting GmbH, Betastr. 13a, 85774 Unterföhring
Geschäftsführer: Henrik Klagges, Christoph Stock, Dr. Robert Dahlke
Sitz: Unterföhring * Amtsgericht München * HRB 135082

> Am 19.01.2016 um 21:26 schrieb Stephan Ewen :
> 
> Hi!
> 
> This kind of error (GC overhead exceeded) usually means that the system is 
> reaching a state where it has very many still living objects and frees little 
> memory during each collection. As a consequence, it is basically busy with 
> only garbage collection.
> 
> Your job probably has about 500-600 MB or free memory, the rest is at that 
> memory size reserved for JVM overhead and Flink's worker memory.
> Now, since your job actually does not keep any objects or rows around, this 
> should be plenty. I can only suspect that the Oracle JDBC driver is very 
> memory hungry, thus pushing the system to the limit. (I found this, for 
> example 
> 
> What you can do:
>  For this kind of job, you can simply tell Flink to not reserve as much 
> memory, by using the option "taskmanager.memory.size=1". If the JDBC driver 
> has no leak, but is simply super hungry, this should solve it.
> 
> Greetings,
> Stephan
> 
> 
> I also found these resources concerning Oracle JDBC memory:
> 
>  - 
> http://stackoverflow.com/questions/2876895/oracle-t4cpreparedstatement-memory-leaks
>  
> 
>  (bottom answer)
>  - https://community.oracle.com/thread/2220078?tstart=0 
> 
> 
> 
> On Tue, Jan 19, 2016 at 5:44 PM, Maximilian Bode  > wrote:
> Hi Robert,
> 
> I am using 0.10.1.
> 
> 
>> Am 19.01.2016 um 17:42 schrieb Robert Metzger > >:
>> 
>> Hi Max,
>> 
>> which version of Flink are you using?
>> 
>> On Tue, Jan 19, 2016 at 5:35 PM, Maximilian Bode 
>> mailto:maximilian.b...@tngtech.com>> wrote:
>> Hi everyone,
>> 
>> I am facing a problem using the JDBCInputFormat which occurred in a larger 
>> Flink job. As a minimal example I can reproduce it when just writing data 
>> into a csv after having read it from a database, i.e.
>> 
>> DataSet> existingData = env.createInput(
>>  JDBCInputFormat.buildJDBCInputFormat()
>>  .setDrivername("oracle.jdbc.driver.OracleDriver")
>>  .setUsername(…)
>>  .setPassword(…)
>>  .setDBUrl(…)
>>  .setQuery("select DATA from TABLENAME")
>>  .finish(),
>>  new TupleTypeInfo<>(Tuple1.class, BasicTypeInfo.STRING_TYPE_INFO));
>> existingData.writeAsCsv(…);
>> 
>> where DATA is a column containing strings of length ~25 characters and 
>> TABLENAME contains 20 million rows.
>> 
>> After starting the job on a YARN cluster (using -tm 3072 and leaving the 
>> other memory settings at default values), Flink happily goes along at first 
>> but then fails after something like three million records have been sent by 
>> the JDBCInputFormat. The Exception reads "The slot in which the task was 
>> executed has been released. Probably loss of TaskManager …". The local 
>> taskmanager.log in the affected container reads
>> "java.lang.OutOfMemoryError: GC overhead limit exceeded
>> at 
>> java.util.Collections$UnmodifiableCollection.iterator(Collections.java:1063)
>> at 
>> org.jboss.netty.channel.socket.nio.NioClientBoss.processConnectTimeout(NioClientBoss.java:119)
>> at 
>> org.jboss.netty.channel.socket.nio.NioClientBoss.process(NioClientBoss.java:83)
>> at 
>> org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:312)
>> at 
>> org.jboss.netty.channel.socket.nio.NioClientBoss.run(NioClientBoss.java:42)
>> at 
>> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
>> at 
>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
>> at java.lang.Thread.run(Thread.java:744)"
>> 
>> Any ideas what is going wrong here?
>> 
>> Cheers,
>> Max
>> 
>> — 
>> Maximilian Bode * Junior Consultant * maximilian.b...@tngtech.com 
>> 

Re: Unexpected out of bounds error in UnilateralSortMerger

[Theodore Vasiloudis]

OK here's what I tried:

* Build Flink (mvn clean install) from the branch you linked (kryo)
* Build my uber-jar, I use SBT with 1.0-SNAPSHOT as the Flink version,
added local maven repo to resolvers so that it picks up the previously
installed version (I hope)
* Launch local cluster from newly built Flink, try to run job

Still getting the same error.

Is there a way to ensure that SBT is picking up the local version of Flink
to build the uber-jar?
Does it matter in this case, or is it enough that I'm sure the launched
Flink instance comes from the branch you linked?

On Wed, Jan 20, 2016 at 4:30 PM, Stephan Ewen  wrote:

> The bug looks to be in the serialization via Kryo while spilling windows.
> Note that Kryo is here used as a fallback serializer, since the
> SparseVector is not transparent type to Flink.
>
> I think there are two possible reasons:
>   1) Kryo, or our Kryo setup has an issue here
>   2) Kryo is inconsistently configured. There are multiple Kryo instances
> used across the serializers in the sorter. There may be a bug that they are
> not initialized in sync.
>
>
> To check this, can you build Flink with this pull request (
> https://github.com/apache/flink/pull/1528) or from this branch (
> https://github.com/StephanEwen/incubator-flink kryo) and see if that
> fixes it?
>
>
> Thanks,
> Stephan
>
>
>
>
>
> On Wed, Jan 20, 2016 at 3:33 PM, Theodore Vasiloudis <
> theodoros.vasilou...@gmail.com> wrote:
>
>> I haven't been able to reproduce this with other datasets. Taking a
>> smaller sample from the large dataset I'm using (link to data
>> )
>> causes the same problem however.
>>
>> I'm wondering if the implementation of readLibSVM is what's wrong here.
>> I've tried the new version commited recently by Chiwan, but I still get the
>> same error.
>>
>> I'll see if I can spot a bug in readLibSVM.
>>
>> On Wed, Jan 20, 2016 at 1:43 PM, Theodore Vasiloudis <
>> theodoros.vasilou...@gmail.com> wrote:
>>
>>> It's on 0.10.
>>>
>>> I've tried explicitly registering SparseVector (which is done anyway by
>>> registerFlinkMLTypes
>>> 
>>> which is called when the SVM predict or evaluate functions are called
>>> )
>>> in my job but I still get the same. I will try a couple different datasets
>>> and try to see if it's the number of features that is causing this or
>>> something else.
>>>
>>> So far it works fine for a dataset with 8 features, but the large one
>>> has 2000 and I get the above error there. I will try large datasets with a
>>> few features and small datasets with many features as well.
>>>
>>> On Wed, Jan 20, 2016 at 11:39 AM, Stephan Ewen  wrote:
>>>
 Hi!

 Does this error occur in 0.10 or im 1.0-SNAPSHOT?

 It is probably an incorrectly configured Kryo instance (not a problem
 of the sorter).
 What is strange is that it occurs in the "MapReferenceResolver" -
 there should be no reference resolution during serialization /
 deserialization.

 Can you try what happens when you explicitly register the type
 SparseVector at the ExecutionEnvironment?

 Stephan


 On Wed, Jan 20, 2016 at 11:24 AM, Theodore Vasiloudis <
 theodoros.vasilou...@gmail.com> wrote:

> Hello all,
>
> I'm trying to run a job using FlinkML and I'm confused about the
> source of an error.
>
> The job reads a libSVM formatted file and trains an SVM classifier on
> it.
>
> I've tried this with small datasets and everything works out fine.
>
> When trying to run the same job on a large dataset (~11GB
> uncompressed) however, I get the following error:
>
>
>> java.lang.RuntimeException: Error obtaining the sorted input: Thread
>> 'SortMerger spilling thread' terminated due to an exception:
>> java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
>> Serialization trace:
>> indices (org.apache.flink.ml.math.SparseVector)
>> at
>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.getIterator(UnilateralSortMerger.java:619)
>> at
>> org.apache.flink.runtime.operators.BatchTask.getInput(BatchTask.java:1089)
>> at
>> org.apache.flink.runtime.operators.NoOpDriver.run(NoOpDriver.java:78)
>> at
>> org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:489)
>> at
>> org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:354)
>> at
>> org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
>> at java.lang.Thread.run(Thread.java:745)
>

Re: Cannot start FlinkMiniCluster.WebServer using different port in FlinkMiniCluster

[Till Rohrmann]

It seems that the web server could not been instantiated. The reason for
this problem should be in your logs. Could you look it up and post the
reason here?

Additionally, we should build in a sanity check to avoid the NPE.

Cheers,
Till

On Wed, Jan 20, 2016 at 5:06 PM, HungChang 
wrote:

> The original port is used so I'm changing the web port but it fails to.
> Can I
> ask which part I made a mistake?
>
> The error:
>
> Exception in thread "main" java.lang.NullPointerException
> at
>
> org.apache.flink.runtime.minicluster.FlinkMiniCluster.startWebServer(FlinkMiniCluster.scala:295)
> at
>
> org.apache.flink.runtime.minicluster.FlinkMiniCluster.start(FlinkMiniCluster.scala:268)
> at
>
> org.apache.flink.runtime.minicluster.FlinkMiniCluster.start(FlinkMiniCluster.scala:226)
> at
>
> org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:101)
> at
>
> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1181)
>
> The code that changing the configuration of execution environment.
>
> Configuration env_config = new Configuration();
> env_config.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true);
>
> env_config.setString(ConfigConstants.JOB_MANAGER_WEB_PORT_KEY, "");
>
> env_config.setString(ConfigConstants.JOB_MANAGER_WEB_LOG_PATH_KEY,
> "./data/dummyLogFile.txt");
>
> StreamExecutionEnvironment env = new
> LocalStreamEnvironment(env_config);
>
>/* Some code here
>   */
>
> env.execute();
>
> Flink version is 0.10.1.
>
>
>
> --
> View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Cannot-start-FlinkMiniCluster-WebServer-using-different-port-in-FlinkMiniCluster-tp4414.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>

Cannot start FlinkMiniCluster.WebServer using different port in FlinkMiniCluster

[HungChang]

The original port is used so I'm changing the web port but it fails to. Can I
ask which part I made a mistake?

The error:

Exception in thread "main" java.lang.NullPointerException
at
org.apache.flink.runtime.minicluster.FlinkMiniCluster.startWebServer(FlinkMiniCluster.scala:295)
at
org.apache.flink.runtime.minicluster.FlinkMiniCluster.start(FlinkMiniCluster.scala:268)
at
org.apache.flink.runtime.minicluster.FlinkMiniCluster.start(FlinkMiniCluster.scala:226)
at
org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:101)
at
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1181)

The code that changing the configuration of execution environment.

Configuration env_config = new Configuration();
env_config.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true);

env_config.setString(ConfigConstants.JOB_MANAGER_WEB_PORT_KEY, "");

env_config.setString(ConfigConstants.JOB_MANAGER_WEB_LOG_PATH_KEY,
"./data/dummyLogFile.txt");

StreamExecutionEnvironment env = new LocalStreamEnvironment(env_config);

   /* Some code here
  */

env.execute();

Flink version is 0.10.1.



--
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Cannot-start-FlinkMiniCluster-WebServer-using-different-port-in-FlinkMiniCluster-tp4414.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.

An interesting apache project: Reef

[kovas boguta]

Some people here (especially Flink contributors) might be interested to
know about this project: https://reef.apache.org/index.html

It is lower-level than Flink (and less mature), but with similar
architectural sensibilities and emphasis on interfaces. It would be pretty
interesting to compare the designs in areas of overlap, and maybe some
cross-pollination of ideas (if not outright reuse/compatibility) is
possible.

Also check out the related subprojects:
https://reef.apache.org/tang.html
https://reef.apache.org/wake.html

Re: flink 1.0-SNAPSHOT scala 2.11 compilation error

[David Kim]

Hi Robert,

Thanks for following up. The issue is resolved!

Cheers,
David

On Wed, Jan 20, 2016 at 3:08 AM, Robert Metzger  wrote:

> Hey David,
>
> the issue should be resolved now. Please let me know if its still an issue
> for you.
>
> Regards,
> Robert
>
>
> On Fri, Jan 15, 2016 at 4:02 PM, David Kim <
> david@braintreepayments.com> wrote:
>
>> Thanks Till! I'll keep an eye out on the JIRA issue. Many thanks for the
>> prompt reply.
>>
>> Cheers,
>> David
>>
>> On Fri, Jan 15, 2016 at 4:16 AM, Till Rohrmann 
>> wrote:
>>
>>> Hi David,
>>>
>>> this is definitely an error on our side which might be caused by the
>>> latest changes to the project structure (removing flink-staging directory).
>>> I’ve filed a JIRA issue https://issues.apache.org/jira/browse/FLINK-3241.
>>> It should be fixed soon.
>>>
>>> In the meantime it should work if you build yourself the Flink binaries
>>> from the sources. First check them out, then call 
>>> tools/change-scala-version.sh
>>> 2.11 in the root directory and then mvn clean install -DskipTests
>>> -Dmaven.javadoc.skip=true. These binaries should depend on the right
>>> Scala version.
>>>
>>> Cheers,
>>> Till
>>> ​
>>>
>>> On Thu, Jan 14, 2016 at 11:25 PM, David Kim <
>>> david@braintreepayments.com> wrote:
>>>
 Hi,

 I have a scala project depending on flink scala_2.11 and am seeing a
 compilation error when using sbt.

 I'm using flink 1.0-SNAPSHOT and my build was working yesterday. I was
 wondering if maybe a recent change to flink could be the cause?

 Usually we see flink resolving the scala _2.11 counterparts for akka
 and scalatest:

 [info] Resolving com.typesafe.akka#akka-actor_2.11;2.3.7 ...
 [info] Resolving com.typesafe#config;1.2.1 ...
 [info] Resolving com.typesafe.akka#akka-remote_2.11;2.3.7 ...[info] 
 Resolving org.scalatest#scalatest_2.11;2.2.4 ...



 but am seeing it pick up the _2.10 counterparts:

 [info] Resolving com.typesafe.akka#akka-actor_2.10;2.3.7 ...[info] 
 Resolving com.typesafe.akka#akka-actor_2.10;2.3.7 ...
 [info] Resolving com.typesafe#config;1.2.1 ...[info] Resolving 
 org.scalatest#scalatest_2.10;2.2.2 ...



 This ultimately gives us the following compilation error:

 [error]org.scalatest:scalatest _2.10, _2.11
 java.lang.RuntimeException: Conflicting cross-version suffixes in: 
 org.scalatest:scalatest
at scala.sys.package$.error(package.scala:27)
at sbt.ConflictWarning$.processCrossVersioned(ConflictWarning.scala:46)
at sbt.ConflictWarning$.apply(ConflictWarning.scala:32)



 Maybe the latest 1.0-SNAPSHOT build erroneously built with scala 2.10
 for the 2.11 profile? Any guidance appreciated!

 Thanks,
 David


>>>
>>
>>
>> --
>> Note: this information is confidential. It is prohibited to share, post
>> online or otherwise publicize without Braintree's prior written consent.
>>
>
>


-- 
Note: this information is confidential. It is prohibited to share, post
online or otherwise publicize without Braintree's prior written consent.

Re: Unexpected out of bounds error in UnilateralSortMerger

[Stephan Ewen]

The bug looks to be in the serialization via Kryo while spilling windows.
Note that Kryo is here used as a fallback serializer, since the
SparseVector is not transparent type to Flink.

I think there are two possible reasons:
  1) Kryo, or our Kryo setup has an issue here
  2) Kryo is inconsistently configured. There are multiple Kryo instances
used across the serializers in the sorter. There may be a bug that they are
not initialized in sync.


To check this, can you build Flink with this pull request (
https://github.com/apache/flink/pull/1528) or from this branch (
https://github.com/StephanEwen/incubator-flink kryo) and see if that fixes
it?


Thanks,
Stephan





On Wed, Jan 20, 2016 at 3:33 PM, Theodore Vasiloudis <
theodoros.vasilou...@gmail.com> wrote:

> I haven't been able to reproduce this with other datasets. Taking a
> smaller sample from the large dataset I'm using (link to data
> )
> causes the same problem however.
>
> I'm wondering if the implementation of readLibSVM is what's wrong here.
> I've tried the new version commited recently by Chiwan, but I still get the
> same error.
>
> I'll see if I can spot a bug in readLibSVM.
>
> On Wed, Jan 20, 2016 at 1:43 PM, Theodore Vasiloudis <
> theodoros.vasilou...@gmail.com> wrote:
>
>> It's on 0.10.
>>
>> I've tried explicitly registering SparseVector (which is done anyway by
>> registerFlinkMLTypes
>> 
>> which is called when the SVM predict or evaluate functions are called
>> )
>> in my job but I still get the same. I will try a couple different datasets
>> and try to see if it's the number of features that is causing this or
>> something else.
>>
>> So far it works fine for a dataset with 8 features, but the large one has
>> 2000 and I get the above error there. I will try large datasets with a few
>> features and small datasets with many features as well.
>>
>> On Wed, Jan 20, 2016 at 11:39 AM, Stephan Ewen  wrote:
>>
>>> Hi!
>>>
>>> Does this error occur in 0.10 or im 1.0-SNAPSHOT?
>>>
>>> It is probably an incorrectly configured Kryo instance (not a problem of
>>> the sorter).
>>> What is strange is that it occurs in the "MapReferenceResolver" - there
>>> should be no reference resolution during serialization / deserialization.
>>>
>>> Can you try what happens when you explicitly register the type
>>> SparseVector at the ExecutionEnvironment?
>>>
>>> Stephan
>>>
>>>
>>> On Wed, Jan 20, 2016 at 11:24 AM, Theodore Vasiloudis <
>>> theodoros.vasilou...@gmail.com> wrote:
>>>
 Hello all,

 I'm trying to run a job using FlinkML and I'm confused about the source
 of an error.

 The job reads a libSVM formatted file and trains an SVM classifier on
 it.

 I've tried this with small datasets and everything works out fine.

 When trying to run the same job on a large dataset (~11GB uncompressed)
 however, I get the following error:


> java.lang.RuntimeException: Error obtaining the sorted input: Thread
> 'SortMerger spilling thread' terminated due to an exception:
> java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
> Serialization trace:
> indices (org.apache.flink.ml.math.SparseVector)
> at
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.getIterator(UnilateralSortMerger.java:619)
> at
> org.apache.flink.runtime.operators.BatchTask.getInput(BatchTask.java:1089)
> at
> org.apache.flink.runtime.operators.NoOpDriver.run(NoOpDriver.java:78)
> at
> org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:489)
> at
> org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:354)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
> at java.lang.Thread.run(Thread.java:745)
> Caused by: java.io.IOException: Thread 'SortMerger spilling thread'
> terminated due to an exception: java.lang.IndexOutOfBoundsException: 
> Index:
> 14, Size: 2
> Serialization trace:
> indices (org.apache.flink.ml.math.SparseVector)
> at
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:800)
> Caused by: com.esotericsoftware.kryo.KryoException:
> java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
> Serialization trace:
> indices (org.apache.flink.ml.math.SparseVector)
> at
> com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125)
> at
> com.esotericsoftware.kryo.serializers.FieldSerializ

Re: Compile fails with scala 2.11.4

[Ritesh Kumar Singh]

Thanks for the update Robert, I tried it out and it works fine for
scala_2.11.4 version.
I've made a docker image
 of the
same and put it up on the hub just in case anyone else needs it.

Thanks,

*Ritesh Kumar Singh,*
*https://riteshtoday.wordpress.com/* 

On Wed, Jan 20, 2016 at 10:07 AM, Robert Metzger 
wrote:

> Hi,
>
> in the latest master, the "tools/change-scala-version.sh" should be
> fixed. Also, the 1.0-SNAPSHOT version deployed to the snapshot repository
> should be good again.
>
> @Ritesh: The commands were correct. I'm not sure if Flink builds with
> Scala 2.11.4, the default 2.11 version we are using is 2.11.7.
>
>
>
> On Tue, Jan 19, 2016 at 7:41 AM, Prez Cannady <
> revp...@correlatesystems.com> wrote:
>
>> Assuming you haven’t already migrated back to 2.10, you might try this;
>> $ git checkout release-0.10
>> $ tools/change-scala-version 2.11
>> $ mvn clean install -DskipTests=true -Dmaven.javadoc.skip=true
>> -Dscala.version=2.11.4 -Dscala.binary.version=2.11
>>
>> Then try building your project.  Building under these conditions resolved
>> my issue which emitted the same error.
>>
>> Prez Cannady
>> p: 617 500 3378
>> e: revp...@opencorrelate.org
>> GH: https://github.com/opencorrelate
>> LI: https://www.linkedin.com/in/revprez
>>
>>
>>
>>
>>
>>
>>
>>
>>
>> On Jan 18, 2016, at 10:20 AM, Ritesh Kumar Singh <
>> riteshoneinamill...@gmail.com> wrote:
>>
>> Thanks for the replies.
>>
>> @Chiwan, I am switching back to scala_2.10.4 for the time being. I was
>> using scala_2.11.4 as this is the version I've compiled spark with. But
>> anyways, I can wait for the bug to be resolved.
>>
>> @Robert, the commands were as follows:
>> $tools/change-scala-version.sh 2.11
>> $mvn clean install -DskipTests -Dscala.version=2.11.4
>>
>> I hope I'm doing it right ?
>>
>> Thanks,
>>
>> *Ritesh Kumar Singh,*
>> *https://riteshtoday.wordpress.com/* 
>>
>> On Mon, Jan 18, 2016 at 12:03 PM, Robert Metzger 
>> wrote:
>>
>>> How did start the Flink for Scala 2.11 compilation ?
>>>
>>> On Mon, Jan 18, 2016 at 11:41 AM, Chiwan Park 
>>> wrote:
>>>
 Hi Ritesh,

 This problem seems already reported [1]. Flink community is
 investigating this issue. I think that if you don’t need Scala 2.11, use
 Scala 2.10 until the issue is solved.

 [1]:
 http://mail-archives.apache.org/mod_mbox/flink-user/201601.mbox/%3CCAB6CeiZ_2snN-piXzd3gHnyQePu_PA0Ro7qXUF8%3DVTxoyL0YyA%40mail.gmail.com%3E

 > On Jan 18, 2016, at 7:24 PM, Ritesh Kumar Singh <
 riteshoneinamill...@gmail.com> wrote:
 >
 > [ERROR]
 /home/flink/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala:703:
 error: can't expand macros compiled by previous versions of Scala
 > [ERROR]   assert(cachedGraph2.isArchived)
 > [ERROR]   ^
 > [ERROR] one error found
 > [INFO]
 
 > [INFO] Reactor Summary:
 > [INFO]
 > [INFO] flink .. SUCCESS [
 24.820 s]
 > [INFO] flink-annotations .. SUCCESS
 [  2.755 s]
 > [INFO] flink-shaded-hadoop  SUCCESS
 [  0.208 s]
 > [INFO] flink-shaded-hadoop2 ... SUCCESS [
 15.627 s]
 > [INFO] flink-shaded-include-yarn-tests  SUCCESS [
 17.076 s]
 > [INFO] flink-shaded-curator ... SUCCESS
 [  0.200 s]
 > [INFO] flink-shaded-curator-recipes ... SUCCESS
 [  2.751 s]
 > [INFO] flink-shaded-curator-test .. SUCCESS
 [  0.355 s]
 > [INFO] flink-core . SUCCESS [
 33.052 s]
 > [INFO] flink-java . SUCCESS [
 10.224 s]
 > [INFO] flink-runtime .. FAILURE
 [01:23 min]
 > [INFO] flink-optimizer  SKIPPED
 >
 >
 > Any workaround for scala_2.11.4 or do I have to switch back to
 scala_2.10.4 ?
 >
 > Thanks,
 > Ritesh Kumar Singh,
 > https://riteshtoday.wordpress.com/
 >

 Regards,
 Chiwan Park


>>>
>>
>>
>

Re: Results of testing Flink quickstart against 0.10-SNAPSHOT and 1.0-SNAPSHOT (re. Dependency on non-existent org.scalamacros:quasiquotes_2.11:)

[Prez Cannady]

Morning, Robert.

You’re right; the 1.0-SNAPSHOT with fetched binaries issue is resolved now.  
Unfortunately, it now emits the same error as 0.10-SNAPSHOT with fetched 
binaries.  There is a fix for that:

https://github.com/apache/flink/pull/1511 


It’s definitely in the release-0.10 and master branches

https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorSnapshot.java
 

https://github.com/apache/flink/blob/release-0.10/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorSnapshot.java
 


I grabbed the sources jar for the latest 1.0-SNAPSHOT and 0.10-SNAPSHOT:

From 
https://repository.apache.org/content/repositories/snapshots/org/apache/flink/flink-runtime_2.11/0.10-SNAPSHOT/flink-runtime_2.11-0.10-20160114.200924-161-sources.jar
 


/**
 * Gets the Flink (internal) accumulators values.
 * @return the serialized map
 */
public Map> 
deserializeFlinkAccumulators() throws IOException, ClassNotFoundException {
return 
flinkAccumulators.deserializeValue(ClassLoader.getSystemClassLoader());
}

ClassLoader.getSystemClassLoader() should be getClass().getClassLoader().

Not sure why it’s not taking in the build, but there’s the problem.

Prez Cannady  
p: 617 500 3378  
e: revp...@opencorrelate.org   
GH: https://github.com/opencorrelate   
LI: https://www.linkedin.com/in/revprez   









> On Jan 20, 2016, at 8:17 AM, Robert Metzger  wrote:
> 
> Hi Prez,
> 
> thanks a lot for the thorough research you did on this issue. The issue with 
> "1.0-SNAPSHOT with fetched binary dependencies" should be resolved by a fix 
> I've pushed to master yesterday:
> 
> a) The "change-scala-version" script wasn't adopted to the renamed examples 
> directory, that's why it failed renaming the artifacts for _2.11. That's why 
> the maven dependencies ended up being mixed between Scala 2.11 and 2.10.
> https://github.com/apache/flink/commit/8f0c47df092ccdc6028dbd88aed197edcd8945ee#diff-c1ca4095c51fcc58f380c59cfabffc8a
>  
> 
> 
> b) The deployment of the scala 2.11 build failed because the "flink-table" 
> module contained java code in the scala/ directory. Also the code was not 
> compliant with our java checkstyle. 
> https://github.com/apache/flink/commit/018f1fee9c856c66454155043006d6f53d39517a
>  
> 
> 
> I hope with these two fixes, the scala 2.11 problems are resolved for now.
> 
> Regards,
> Robert
> 
> On Tue, Jan 19, 2016 at 6:43 AM, Prez Cannady  > wrote:
> One correction, line 3 of 1.0-SHAPSHOT source build should read “checked out 
> master branch (snapshot version 1.0-SNAPSHOT)."
> 
> Prez Cannady  
> p: 617 500 3378  
> e: revp...@opencorrelate.org   
> GH: https://github.com/opencorrelate   
> LI: https://www.linkedin.com/in/revprez  
>  
> 
> 
> 
> 
> 
> 
> 
> 
> 
>> On Jan 19, 2016, at 12:41 AM, Prez Cannady > > wrote:
>> 
>> Sent this to d...@flink.apache.org , but that 
>> might not be the appropriate forum for it.
>> 
>> Finally got a chance to sit down and run through a few tests. Upfront, I 
>> have been able to resolve my issue sufficiently to move forward, but seems 
>> there’s an issue with the current bits for both 1.0-SNAPSHOT and 
>> 0.10-SNAPSHOT in the remote Maven repos.
>> 
>> Notes
>> 
>> wordcount-processing  is 
>> a customized version of the Flink quickstart archetype I’m using to test 
>> Flink integration with Spring Boot. It is instrumented for Maven and Gradle 
>> build and execution.
>> I’m targeting Scala 2.11 and Flink 0.10.
>> 0.10-SNAPSHOT source build
>> 
>> Steps
>> 
>> Checked out release–0.10 branch (snapshot version 0.10-SNAPSHOT).
>> Built with mvn clean install -DskipTests=true -Dmaven.javadoc.skip=true 
>> -Dscala.version=2.11.7 -Dscala.binary.version=2.11.
>> Ran wordcount-process with mvn clean spring-boot:run 
>> -Drun.arguments=“localhost,”.
>> Ran wordcount-process with gradle bootRun -Drun.arguments=“localhost 

Re: groupBy(int) is undefined for the type SingleOutputStreamOperator while running streaming example provided on webpage

[Robert Metzger]

Hi,

as you can see from the URL, its the documentation for Flink 0.8. The
current documentation is here:
https://ci.apache.org/projects/flink/flink-docs-release-0.10/apis/streaming_guide.html

In the 0.10. release announcement (
http://flink.apache.org/news/2015/11/16/release-0.10.0.html ) the change
has been mentioned.

On Wed, Jan 20, 2016 at 3:47 PM, Vinaya M S  wrote:

> Version FLink 0.10.
>
> Example is mentioned in
> https://ci.apache.org/projects/flink/flink-docs-release-0.8/streaming_guide.html
> Please let me know where I can find these kind of updates.
>
> Thank you.
>
> On Wed, Jan 20, 2016 at 9:41 AM, Robert Metzger 
> wrote:
>
>> Hi.
>>
>> which Flink version are you using?
>> Starting from Flink 0.10., "groupBy" has been renamed to "keyBy".
>>
>> Where did you find the "groupBy" example?
>>
>> On Wed, Jan 20, 2016 at 3:37 PM, Vinaya M S  wrote:
>>
>>> Hi,
>>>
>>> I'm new to Flink. Can anyone help me with the error below.
>>>
>>> Exception in thread "main" java.lang.Error: Unresolved compilation
>>> problem:
>>>
>>> The method groupBy(int) is undefined for the type
>>> SingleOutputStreamOperator,capture#1-of ?>
>>>
>>>
>>> The code snippet is:
>>>
>>> DataStream> dataStream = env
>>>
>>> .socketTextStream("localhost", )
>>>
>>> .flatMap(new Splitter())
>>>
>>> .groupBy(0)
>>>
>>> .sum(1);
>>>
>>>
>>> Thank you.
>>>
>>
>>
>

Re: groupBy(int) is undefined for the type SingleOutputStreamOperator while running streaming example provided on webpage

[Robert Metzger]

Hi.

which Flink version are you using?
Starting from Flink 0.10., "groupBy" has been renamed to "keyBy".

Where did you find the "groupBy" example?

On Wed, Jan 20, 2016 at 3:37 PM, Vinaya M S  wrote:

> Hi,
>
> I'm new to Flink. Can anyone help me with the error below.
>
> Exception in thread "main" java.lang.Error: Unresolved compilation
> problem:
>
> The method groupBy(int) is undefined for the type
> SingleOutputStreamOperator,capture#1-of ?>
>
>
> The code snippet is:
>
> DataStream> dataStream = env
>
> .socketTextStream("localhost", )
>
> .flatMap(new Splitter())
>
> .groupBy(0)
>
> .sum(1);
>
>
> Thank you.
>

Re: parallelism parameter and output relation

[Robert Metzger]

Hi Serkan,

yes, with parallelism=1, you'll get one file, with everything higher, Flink
is creating a directory with a file for each parallel instance.
In your case, Flink can not create (or write to) the file because there is
already a directory with the same name. Can you delete the directory and
see if writing to the file works afterwards?

Regards,
Robert


2016-01-20 12:53 GMT+01:00 Serkan Taş :

> I am working on this example
> http://www.itshared.org/2015/03/naive-bayes-on-apache-flink.html to learn
> get some more experience on platform.
>
> Question is ;
>
> By default, the output of process is double file (named 1 and 2) located
> in created folder. If i set parallelism to 1, FileNotFound exception is
> thrown.
>
> I was expecting to get a single file instead, am i right ?
>
>
> *Serkan Taş*
> Mobil : +90 532 250 07 71
> Likya Bilgi Teknolojileri
> ve İletişim Hiz. Ltd. Şti.
> www.likyateknoloji.com
>
> --
> Bu elektronik posta ve onunla iletilen bütün dosyalar gizlidir. Sadece
> yukarıda isimleri belirtilen kişiler arasında özel haberleşme amacını
> taşımaktadır. Size yanlışlıkla ulaşmışsa bu elektonik postanın içeriğini
> açıklamanız, kopyalamanız, yönlendirmeniz ve kullanmanız kesinlikle
> yasaktır. Lütfen mesajı geri gönderiniz ve sisteminizden siliniz. Likya
> Bilgi Teknolojileri ve İletişim Hiz. Ltd. Şti. bu mesajın içeriği ile
> ilgili olarak hiç bir hukuksal sorumluluğu kabul etmez.
>
> This electronic mail and any files transmitted with it are intended for
> the private use of  the persons named above. If you received this message
> in error, forwarding, copying or use of any of the information is strictly
> prohibited. Please immediately notify the sender and delete it from your
> system. Likya Bilgi Teknolojileri ve İletişim Hiz. Ltd. Şti. does not
> accept legal responsibility for the contents of this message.
> --
>
>
>
>
>
>
>
> P
> Bu e-postayı yazdırmadan önce, çevreye olan sorumluluğunuzu tekrar düşünün.
> Please consider your environmental responsibility before printing this
> e-mail.
>
>
>

Re: groupBy(int) is undefined for the type SingleOutputStreamOperator while running streaming example provided on webpage

[Vinaya M S]

Version FLink 0.10.

Example is mentioned in
https://ci.apache.org/projects/flink/flink-docs-release-0.8/streaming_guide.html
Please let me know where I can find these kind of updates.

Thank you.

On Wed, Jan 20, 2016 at 9:41 AM, Robert Metzger  wrote:

> Hi.
>
> which Flink version are you using?
> Starting from Flink 0.10., "groupBy" has been renamed to "keyBy".
>
> Where did you find the "groupBy" example?
>
> On Wed, Jan 20, 2016 at 3:37 PM, Vinaya M S  wrote:
>
>> Hi,
>>
>> I'm new to Flink. Can anyone help me with the error below.
>>
>> Exception in thread "main" java.lang.Error: Unresolved compilation
>> problem:
>>
>> The method groupBy(int) is undefined for the type
>> SingleOutputStreamOperator,capture#1-of ?>
>>
>>
>> The code snippet is:
>>
>> DataStream> dataStream = env
>>
>> .socketTextStream("localhost", )
>>
>> .flatMap(new Splitter())
>>
>> .groupBy(0)
>>
>> .sum(1);
>>
>>
>> Thank you.
>>
>
>

groupBy(int) is undefined for the type SingleOutputStreamOperator while running streaming example provided on webpage

[Vinaya M S]

Hi,

I'm new to Flink. Can anyone help me with the error below.

Exception in thread "main" java.lang.Error: Unresolved compilation problem:

The method groupBy(int) is undefined for the type
SingleOutputStreamOperator,capture#1-of ?>


The code snippet is:

DataStream> dataStream = env

.socketTextStream("localhost", )

.flatMap(new Splitter())

.groupBy(0)

.sum(1);


Thank you.

Re: Unexpected out of bounds error in UnilateralSortMerger

[Theodore Vasiloudis]

I haven't been able to reproduce this with other datasets. Taking a smaller
sample from the large dataset I'm using (link to data
)
causes the same problem however.

I'm wondering if the implementation of readLibSVM is what's wrong here.
I've tried the new version commited recently by Chiwan, but I still get the
same error.

I'll see if I can spot a bug in readLibSVM.

On Wed, Jan 20, 2016 at 1:43 PM, Theodore Vasiloudis <
theodoros.vasilou...@gmail.com> wrote:

> It's on 0.10.
>
> I've tried explicitly registering SparseVector (which is done anyway by
> registerFlinkMLTypes
> 
> which is called when the SVM predict or evaluate functions are called
> )
> in my job but I still get the same. I will try a couple different datasets
> and try to see if it's the number of features that is causing this or
> something else.
>
> So far it works fine for a dataset with 8 features, but the large one has
> 2000 and I get the above error there. I will try large datasets with a few
> features and small datasets with many features as well.
>
> On Wed, Jan 20, 2016 at 11:39 AM, Stephan Ewen  wrote:
>
>> Hi!
>>
>> Does this error occur in 0.10 or im 1.0-SNAPSHOT?
>>
>> It is probably an incorrectly configured Kryo instance (not a problem of
>> the sorter).
>> What is strange is that it occurs in the "MapReferenceResolver" - there
>> should be no reference resolution during serialization / deserialization.
>>
>> Can you try what happens when you explicitly register the type
>> SparseVector at the ExecutionEnvironment?
>>
>> Stephan
>>
>>
>> On Wed, Jan 20, 2016 at 11:24 AM, Theodore Vasiloudis <
>> theodoros.vasilou...@gmail.com> wrote:
>>
>>> Hello all,
>>>
>>> I'm trying to run a job using FlinkML and I'm confused about the source
>>> of an error.
>>>
>>> The job reads a libSVM formatted file and trains an SVM classifier on it.
>>>
>>> I've tried this with small datasets and everything works out fine.
>>>
>>> When trying to run the same job on a large dataset (~11GB uncompressed)
>>> however, I get the following error:
>>>
>>>
 java.lang.RuntimeException: Error obtaining the sorted input: Thread
 'SortMerger spilling thread' terminated due to an exception:
 java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
 Serialization trace:
 indices (org.apache.flink.ml.math.SparseVector)
 at
 org.apache.flink.runtime.operators.sort.UnilateralSortMerger.getIterator(UnilateralSortMerger.java:619)
 at
 org.apache.flink.runtime.operators.BatchTask.getInput(BatchTask.java:1089)
 at
 org.apache.flink.runtime.operators.NoOpDriver.run(NoOpDriver.java:78)
 at
 org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:489)
 at
 org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:354)
 at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
 at java.lang.Thread.run(Thread.java:745)
 Caused by: java.io.IOException: Thread 'SortMerger spilling thread'
 terminated due to an exception: java.lang.IndexOutOfBoundsException: Index:
 14, Size: 2
 Serialization trace:
 indices (org.apache.flink.ml.math.SparseVector)
 at
 org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:800)
 Caused by: com.esotericsoftware.kryo.KryoException:
 java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
 Serialization trace:
 indices (org.apache.flink.ml.math.SparseVector)
 at
 com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125)
 at
 com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
 at
 com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
 at
 org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:222)
 at
 org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:236)
 at
 org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:246)
 at
 org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase.copy(TupleSerializerBase.java:73)
 at
 org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase.copy(TupleSerializerBase.java:73)
 at
 org.apache.flink.runtime.operators.sort.NormalizedKeySorter.writeToOutput(NormalizedKeySorter.java:499)
 at
 org.apache.flink.runtime.

Re: Could not upload the jar files to the job manager IOException

[Robert Metzger]

Hi,

can you check the log file of the JobManager you're trying to submit the
job to?
Maybe there you can find helpful information why it failed.

On Wed, Jan 20, 2016 at 3:23 PM, Ana M. Martinez  wrote:

> Hi all,
>
> I am running some experiments with flink in an Amazon cluster and every
> now and then (it seems to appear at random) I get the following IOException:
> > org.apache.flink.client.program.ProgramInvocationException: The program
> execution failed: Could not upload the jar files to the job manager.
>
> Sometimes when it fails, I just try to run it again immediately afterwords
> and it works fine. Any idea on why that might be happening?
>
> Thanks,
> Ana

Could not upload the jar files to the job manager IOException

[Ana M. Martinez]

Hi all,

I am running some experiments with flink in an Amazon cluster and every now and 
then (it seems to appear at random) I get the following IOException:
> org.apache.flink.client.program.ProgramInvocationException: The program 
> execution failed: Could not upload the jar files to the job manager.

Sometimes when it fails, I just try to run it again immediately afterwords and 
it works fine. Any idea on why that might be happening?

Thanks,
Ana

Re: Reading Binary Data (Matrix) with Flink

[Saliya Ekanayake]

Thank you, I saw the readHadoopFile, but I was not sure how it can be used
to the following, which is what I need. The logic of the code requires an
entire row to operate on, so in our current implementation with P tasks,
each of them will read a rectangular block of (N/P) x N from the matrix. Is
this possible with readHadoopFile? Also, the file may not be in hdfs, so is
it possible to refer to local disk in doing this?

Thank you

On Wed, Jan 20, 2016 at 1:31 AM, Chiwan Park  wrote:

> Hi Saliya,
>
> You can use the input format from Hadoop in Flink by using readHadoopFile
> method. The method returns a dataset which of type is Tuple2.
> Note that MapReduce equivalent transformation in Flink is composed of map,
> groupBy, and reduceGroup.
>
> > On Jan 20, 2016, at 3:04 PM, Suneel Marthi  wrote:
> >
> > Guess u r looking for Flink's BinaryInputFormat to be able to read
> blocks of data from HDFS
> >
> >
> https://ci.apache.org/projects/flink/flink-docs-release-0.10/api/java/org/apache/flink/api/common/io/BinaryInputFormat.html
> >
> > On Wed, Jan 20, 2016 at 12:45 AM, Saliya Ekanayake 
> wrote:
> > Hi,
> >
> > I am trying to use Flink perform a parallel batch operation on a NxN
> matrix represented as a binary file. Each (i,j) element is stored as a Java
> Short value. In a typical MapReduce programming with Hadoop, each map task
> will read a block of rows of this matrix and perform computation on that
> block and emit result to the reducer.
> >
> > How is this done in Flink? I am new to Flink and couldn't find a binary
> reader so far. Any help is greatly appreciated.
> >
> > Thank you,
> > Saliya
> >
> > --
> > Saliya Ekanayake
> > Ph.D. Candidate | Research Assistant
> > School of Informatics and Computing | Digital Science Center
> > Indiana University, Bloomington
> > Cell 812-391-4914
> > http://saliya.org
> >
>
> Regards,
> Chiwan Park
>
>


-- 
Saliya Ekanayake
Ph.D. Candidate | Research Assistant
School of Informatics and Computing | Digital Science Center
Indiana University, Bloomington
Cell 812-391-4914
http://saliya.org

Re: integration with a scheduler

[Robert Metzger]

Hi Serkan,

I would suggest to have a look at the "./bin/flink" tool. It allows you to
start ("run") and stop ("cancel") batch and streaming jobs.
Flink doesn't support suspending jobs.

You can also use the JobManager web interface (default port: 8081) to get
the status of the job and also to cancel it.

If you need programmatic access to the job status, have a look at the
CliFrontend class. Its the class called by the "./bin/flink" tool. There,
you can see how jobs are submitted and cancelled.

Regards,
Robert


2016-01-19 12:10 GMT+01:00 :

> Hi,
>
> I am planning to integrate flink with our job scheduler product to execute
> jobs - especially bathc like - on flink which may be the part of some
> other DAG style job chain.
>
> I need some control ablities like start, stop, suspend, get status...
>
> Where shold i go through ?
>
> --
> Serkan Tas
> Likya Bilgi Teknolojileri
> ve Iletişim Hiz. Ltd.
> www.likyateknoloji.com
> Tel : 0 216 471 81 55
> Gsm : 0 542 242 00 92
> Faks:  0 216 661 14 92
>
> --
> Bu elektronik posta ve onunla iletilen bütün dosyalar gizlidir. Sadece
> yukarıda isimleri belirtilen kişiler arasında özel haberleşme amacını
> taşımaktadır. Size yanlışlıkla ulaşmışsa bu elektonik postanın içeriğini
> açıklamanız, kopyalamanız, yönlendirmeniz ve kullanmanız kesinlikle
> yasaktır. Lütfen mesajı geri gönderiniz ve sisteminizden siliniz. Likya
> Bilgi Teknolojileri ve İletişim Hiz. Ltd. Şti. bu mesajın içeriği ile
> ilgili olarak hiç bir hukuksal sorumluluğu kabul etmez.
>
> This electonic mail and any files transmitted with it are intended for the
> private use of  the persons named above. If you received this message in
> error, forwarding, copying or use of any of the information is strictly
> prohibited. Please immediately notify the sender and delete it from your
> system. Likya Bilgi Teknolojileri ve İletişim Hiz. Ltd. Şti. does not
> accept legal responsibility for the contents of this message.
> --
>
> 
> Bu e-postayı yazdırmadan önce, çevreye olan sorumluluğunuzu tekrar düşünün.
> Please consider your environmental responsibility before printing this
> e-mail.
>
>
>
>

Re: Actual byte-streams in multiple-node pipelines

[Tal Maoz]

Hey Robert,

Thanks for responding!

The latency I'm talking about would be no more than 1 second from input to
output (meaning, bytes should flow immediately through the pipline and get
to the other side after going through the processing). You can assume the
processors have enough power to work in real-time.

The processors would be, for the most part, running external processes
(binary executables) and will feed them the incoming data, and then pass
along their stdout to the next stage. Simply put, I have several existing
'blackbox' utilities that I need to run on the data in sequence and each of
which is a CPU hog...

Regarding fault tolerance, no data should be lost and each processor should
get the data ONCE and in the correct order (when data is supposed to flow
to the same processor). If a node crashes, a new one will take it's place
and the data that was sent to the crashed node and was not processed should
be sent to the new one, while the output should flow transparently to the
next node as if no crashes happened. I know this is a very complicated
demand but it is a must in my case.

Finally, I'm talking about multiple pipelines running, where each node in a
pipeline will be pre-configured before data starts flowing. Each pipeline
will read data from a socket or from an MQ if such an MQ exists and is able
to handle the load with the required low-latency. Each pipeline's source
could be at the range of 45-600MB/s (this can't be split into multiple
sources) and eventually, with enough resources and scaling, the system
should support hundreds of such pipelines, each with it's own source! Also,
at some point, 2 or more sources could be joined with some transformation
into a single data stream... Assume the network fabric itself is capable of
moving those amounts of data...

If i use DataStream where i divide a single segment into very small
buffers for low-latency, how can ensure that, on the one hand the data for
a single segments flows entirely to the same processor while on the other,
different segments can be balanced between several processors?


Tal

On Wed, Jan 20, 2016 at 3:02 PM, Robert Metzger  wrote:

> Hi Tal,
>
> that sounds like an interesting use case. I think I need a bit more
> details about your use case to see how it can be done with Flink.
> You said you need low latency, what latency is acceptable for you?
>
> Also, I was wondering how are you going to feed the input data into Flink?
> If the data is coming from multiple sources, maybe everything can be done
> completely parallel.
> Do you need any fault tolerance guarantees?
>
> You can use Flink's DataStream abstraction with different data types, and
> you could create a DataStream. Flink would internally still send
> multiple of those records in one buffer.
> I think the more efficient approach is, as you suggested, to use a
> DataStream of larger chunks.
>
> What kind of transformations are you planning to do on the stream?
>
> Regarding the amount of data we are talking about here: Flink is certainly
> able to handle those loads. I recently did some tests with our
> KafkaConsumer and I was able to read 390 megabytes/second on my laptop,
> using a parallelism of one (so only one reading thread). My SSD has a read
> rate of 530 MBs/. With sufficiently fast hardware, a few Flink TaskManagers
> will be able to read 600MB/s.
>
>
>
> On Wed, Jan 20, 2016 at 1:39 PM, Ritesh Kumar Singh <
> riteshoneinamill...@gmail.com> wrote:
>
>> I think with sufficient processing power flink can do the above mentioned
>> task using the stream api
>> 
>> .
>>
>> Thanks,
>> *Ritesh Kumar Singh,*
>> *https://riteshtoday.wordpress.com/* 
>>
>> On Wed, Jan 20, 2016 at 11:18 AM, Tal Maoz  wrote:
>>
>>> Hey,
>>>
>>>
>>>
>>> I’m a new user to Flink and I’m trying to figure out if I can build a
>>> pipeline I’m working on using Flink.
>>>
>>> I have a data source that sends out a continues data stream at a
>>> bandwidth of anywhere between 45MB/s to 600MB/s (yes, that’s MiB/s, not
>>> Mib/s, and NOT a series of individual messages but an actual continues
>>> stream of data where some data may depend on previous or future data to be
>>> fully deciphered).
>>>
>>> I need to be able to pass the data through several processing stages
>>> (that manipulate the data but still produce the same order of magnitude
>>> output at each stage) and I need processing to be done with low-latency.
>>>
>>> The data itself CAN be segmented but the segments will be some HUGE
>>> (~100MB – 250MB) and I would like to be able to stream data in and out of
>>> the processors ASAP instead of waiting for full segments to be complete at
>>> each stage (so bytes will flow in/out as soon as they are available).
>>>
>>>
>>>
>>> The obvious solution would be to split the data into very small buffers,
>>> but since each segment would have to be sent completely to the sam

Re: Results of testing Flink quickstart against 0.10-SNAPSHOT and 1.0-SNAPSHOT (re. Dependency on non-existent org.scalamacros:quasiquotes_2.11:)

[Robert Metzger]

Hi Prez,

thanks a lot for the thorough research you did on this issue. The issue
with "1.0-SNAPSHOT with fetched binary dependencies" should be resolved by
a fix I've pushed to master yesterday:

a) The "change-scala-version" script wasn't adopted to the renamed examples
directory, that's why it failed renaming the artifacts for _2.11. That's
why the maven dependencies ended up being mixed between Scala 2.11 and 2.10.
https://github.com/apache/flink/commit/8f0c47df092ccdc6028dbd88aed197edcd8945ee#diff-c1ca4095c51fcc58f380c59cfabffc8a

b) The deployment of the scala 2.11 build failed because the "flink-table"
module contained java code in the scala/ directory. Also the code was not
compliant with our java checkstyle.
https://github.com/apache/flink/commit/018f1fee9c856c66454155043006d6f53d39517a

I hope with these two fixes, the scala 2.11 problems are resolved for now.

Regards,
Robert

On Tue, Jan 19, 2016 at 6:43 AM, Prez Cannady 
wrote:

> One correction, line 3 of *1.0-SHAPSHOT source build* should read “checked
> out *master* branch (snapshot version *1.0-SNAPSHOT*)."
>
> Prez Cannady
> p: 617 500 3378
> e: revp...@opencorrelate.org
> GH: https://github.com/opencorrelate
> LI: https://www.linkedin.com/in/revprez
>
>
>
>
>
>
>
>
>
> On Jan 19, 2016, at 12:41 AM, Prez Cannady 
> wrote:
>
> Sent this to d...@flink.apache.org, but that might not be the appropriate
> forum for it.
>
> Finally got a chance to sit down and run through a few tests. Upfront, I
> have been able to resolve my issue sufficiently to move forward, but seems
> there’s an issue with the current bits for both 1.0-SNAPSHOT and
> 0.10-SNAPSHOT in the remote Maven repos.
>
> *Notes*
>
>1. wordcount-processing
> is a customized
>version of the Flink quickstart archetype I’m using to test Flink
>integration with Spring Boot. It is instrumented for Maven and Gradle build
>and execution.
>2. I’m targeting Scala 2.11 and Flink 0.10.
>
> *0.10-SNAPSHOT source build*
>
> *Steps*
>
>1. Checked out *release–0.10* branch (snapshot version *0.10-SNAPSHOT*
>).
>2. Built with mvn clean install -DskipTests=true
>-Dmaven.javadoc.skip=true -Dscala.version=2.11.7 
> -Dscala.binary.version=2.11
>.
>3. Ran wordcount-process with mvn clean spring-boot:run
>-Drun.arguments=“localhost,”.
>4. Ran wordcount-process with gradle bootRun
>-Drun.arguments=“localhost ”.
>
> *Result*
>
>1. Maven execution of test succeeds without incident.
>2. Gradle execution of test succeeds without incident.
>
> *0.10-SNAPSHOT with fetched binary dependencies*
>
> *Steps*
>
>1. Cleaned out local maven repository with rm -rf
>$HOME/.m2/repo/org/apache/flink && rm -rf $HOME/.m2/repo/org/scala*.
>2. Cleaned out local gradle repository with rm -rf
>$HOME/.gradle/caches/modules-2/files-2.1/org.apache.flink && rm -rf
>$HOME/.gradle/caches/modules-2/files-2.1/org.scala*.
>3. Ran wordcount-process with mvn clean spring-boot:run
>-Drun.arguments=“localhost,”.
>4. Ran wordcount-process with gradle bootRun
>-Drun.arguments=“localhost ”.
>
> *Result*
>
>1. Maven build completed without incident. Maven execution error’d out
>with issue supposedly resolved with pull request 1511
>.
>2. Gradle execution of test succeeds without incident.
>
> *1.0-SNAPSHOT source build*
>
> *Steps*
>
>1. Cleaned out local maven repository with rm -rf
>$HOME/.m2/repo/org/apache/flink && rm -rf $HOME/.m2/repo/org/scala*.
>2. Cleaned out local gradle repository with rm -rf
>$HOME/.gradle/caches/modules-2/files-2.1/org.apache.flink && rm -rf
>$HOME/.gradle/caches/modules-2/files-2.1/org.scala*.
>3. Checked out *release–0.10* branch (snapshot version *0.10-SNAPSHOT*
>).
>4. Built with mvn clean install -DskipTests=true
>-Dmaven.javadoc.skip=true -Dscala.version=2.11.7 
> -Dscala.binary.version=2.11
>.
>5. Ran wordcount-process with mvn clean spring-boot:run
>-Drun.arguments=“localhost,”.
>6. Ran wordcount-process with gradle bootRun
>-Drun.arguments=“localhost ”.
>
> *Result*
>
>1. Maven build completed without incident. Maven execution error’d out
>with issue supposedly resolved with pull request 1511
>.
>2. Gradle execution of test succeeds without incident.
>
> *1.0-SNAPSHOT with fetched binary dependencies*
>
> *Steps*
>
>1. Cleaned out local maven repository with rm -rf
>$HOME/.m2/repo/org/apache/flink && rm -rf $HOME/.m2/repo/org/scala*.
>2. Cleaned out local gradle repository with rm -rf
>$HOME/.gradle/caches/modules-2/files-2.1/org.apache.flink && rm -rf
>$HOME/.gradle/caches/modules-2/files-2.1/org.scala*.
>3. Ran wordcount-process with mvn clean spring-boot:run
>-Drun.arguments=“localhost,”.
>4. Ran wordcount-process with gr

Re: Actual byte-streams in multiple-node pipelines

[Robert Metzger]

Hi Tal,

that sounds like an interesting use case. I think I need a bit more details
about your use case to see how it can be done with Flink.
You said you need low latency, what latency is acceptable for you?

Also, I was wondering how are you going to feed the input data into Flink?
If the data is coming from multiple sources, maybe everything can be done
completely parallel.
Do you need any fault tolerance guarantees?

You can use Flink's DataStream abstraction with different data types, and
you could create a DataStream. Flink would internally still send
multiple of those records in one buffer.
I think the more efficient approach is, as you suggested, to use a
DataStream of larger chunks.

What kind of transformations are you planning to do on the stream?

Regarding the amount of data we are talking about here: Flink is certainly
able to handle those loads. I recently did some tests with our
KafkaConsumer and I was able to read 390 megabytes/second on my laptop,
using a parallelism of one (so only one reading thread). My SSD has a read
rate of 530 MBs/. With sufficiently fast hardware, a few Flink TaskManagers
will be able to read 600MB/s.



On Wed, Jan 20, 2016 at 1:39 PM, Ritesh Kumar Singh <
riteshoneinamill...@gmail.com> wrote:

> I think with sufficient processing power flink can do the above mentioned
> task using the stream api
> 
> .
>
> Thanks,
> *Ritesh Kumar Singh,*
> *https://riteshtoday.wordpress.com/* 
>
> On Wed, Jan 20, 2016 at 11:18 AM, Tal Maoz  wrote:
>
>> Hey,
>>
>>
>>
>> I’m a new user to Flink and I’m trying to figure out if I can build a
>> pipeline I’m working on using Flink.
>>
>> I have a data source that sends out a continues data stream at a
>> bandwidth of anywhere between 45MB/s to 600MB/s (yes, that’s MiB/s, not
>> Mib/s, and NOT a series of individual messages but an actual continues
>> stream of data where some data may depend on previous or future data to be
>> fully deciphered).
>>
>> I need to be able to pass the data through several processing stages
>> (that manipulate the data but still produce the same order of magnitude
>> output at each stage) and I need processing to be done with low-latency.
>>
>> The data itself CAN be segmented but the segments will be some HUGE
>> (~100MB – 250MB) and I would like to be able to stream data in and out of
>> the processors ASAP instead of waiting for full segments to be complete at
>> each stage (so bytes will flow in/out as soon as they are available).
>>
>>
>>
>> The obvious solution would be to split the data into very small buffers,
>> but since each segment would have to be sent completely to the same
>> processor node (and not split between several nodes), doing such
>> micro-batching would be a bad idea as it would spread a single segment’s
>> buffers between multiple nodes.
>>
>>
>>
>> Is there any way to accomplish this with Flink? Or is Flink the wrong
>> platform for that type of processing?
>>
>>
>>
>> Any help would be greatly appreciated!
>>
>>
>>
>> Thanks,
>>
>>
>>
>> Tal
>>
>
>

Re: Unexpected out of bounds error in UnilateralSortMerger

[Theodore Vasiloudis]

It's on 0.10.

I've tried explicitly registering SparseVector (which is done anyway by
registerFlinkMLTypes

which is called when the SVM predict or evaluate functions are called
)
in my job but I still get the same. I will try a couple different datasets
and try to see if it's the number of features that is causing this or
something else.

So far it works fine for a dataset with 8 features, but the large one has
2000 and I get the above error there. I will try large datasets with a few
features and small datasets with many features as well.

On Wed, Jan 20, 2016 at 11:39 AM, Stephan Ewen  wrote:

> Hi!
>
> Does this error occur in 0.10 or im 1.0-SNAPSHOT?
>
> It is probably an incorrectly configured Kryo instance (not a problem of
> the sorter).
> What is strange is that it occurs in the "MapReferenceResolver" - there
> should be no reference resolution during serialization / deserialization.
>
> Can you try what happens when you explicitly register the type
> SparseVector at the ExecutionEnvironment?
>
> Stephan
>
>
> On Wed, Jan 20, 2016 at 11:24 AM, Theodore Vasiloudis <
> theodoros.vasilou...@gmail.com> wrote:
>
>> Hello all,
>>
>> I'm trying to run a job using FlinkML and I'm confused about the source
>> of an error.
>>
>> The job reads a libSVM formatted file and trains an SVM classifier on it.
>>
>> I've tried this with small datasets and everything works out fine.
>>
>> When trying to run the same job on a large dataset (~11GB uncompressed)
>> however, I get the following error:
>>
>>
>>> java.lang.RuntimeException: Error obtaining the sorted input: Thread
>>> 'SortMerger spilling thread' terminated due to an exception:
>>> java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
>>> Serialization trace:
>>> indices (org.apache.flink.ml.math.SparseVector)
>>> at
>>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.getIterator(UnilateralSortMerger.java:619)
>>> at
>>> org.apache.flink.runtime.operators.BatchTask.getInput(BatchTask.java:1089)
>>> at
>>> org.apache.flink.runtime.operators.NoOpDriver.run(NoOpDriver.java:78)
>>> at
>>> org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:489)
>>> at
>>> org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:354)
>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
>>> at java.lang.Thread.run(Thread.java:745)
>>> Caused by: java.io.IOException: Thread 'SortMerger spilling thread'
>>> terminated due to an exception: java.lang.IndexOutOfBoundsException: Index:
>>> 14, Size: 2
>>> Serialization trace:
>>> indices (org.apache.flink.ml.math.SparseVector)
>>> at
>>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:800)
>>> Caused by: com.esotericsoftware.kryo.KryoException:
>>> java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
>>> Serialization trace:
>>> indices (org.apache.flink.ml.math.SparseVector)
>>> at
>>> com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125)
>>> at
>>> com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
>>> at
>>> com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
>>> at
>>> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:222)
>>> at
>>> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:236)
>>> at
>>> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:246)
>>> at
>>> org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase.copy(TupleSerializerBase.java:73)
>>> at
>>> org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase.copy(TupleSerializerBase.java:73)
>>> at
>>> org.apache.flink.runtime.operators.sort.NormalizedKeySorter.writeToOutput(NormalizedKeySorter.java:499)
>>> at
>>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger$SpillingThread.go(UnilateralSortMerger.java:1344)
>>> at
>>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:796)
>>> Caused by: java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
>>> at java.util.ArrayList.rangeCheck(ArrayList.java:653)
>>> at java.util.ArrayList.set(ArrayList.java:444)
>>> at
>>> com.esotericsoftware.kryo.util.MapReferenceResolver.setReadObject(MapReferenceResolver.java:38)
>>> at com.esotericsoftware.kryo.Kryo.reference(Kryo.java:823)
>>> at com.esotericsoftware.kryo.Kryo.readObjectOr

Re: Actual byte-streams in multiple-node pipelines

[Ritesh Kumar Singh]

I think with sufficient processing power flink can do the above mentioned
task using the stream api

.

Thanks,
*Ritesh Kumar Singh,*
*https://riteshtoday.wordpress.com/* 

On Wed, Jan 20, 2016 at 11:18 AM, Tal Maoz  wrote:

> Hey,
>
>
>
> I’m a new user to Flink and I’m trying to figure out if I can build a
> pipeline I’m working on using Flink.
>
> I have a data source that sends out a continues data stream at a bandwidth
> of anywhere between 45MB/s to 600MB/s (yes, that’s MiB/s, not Mib/s, and
> NOT a series of individual messages but an actual continues stream of data
> where some data may depend on previous or future data to be fully
> deciphered).
>
> I need to be able to pass the data through several processing stages (that
> manipulate the data but still produce the same order of magnitude output at
> each stage) and I need processing to be done with low-latency.
>
> The data itself CAN be segmented but the segments will be some HUGE
> (~100MB – 250MB) and I would like to be able to stream data in and out of
> the processors ASAP instead of waiting for full segments to be complete at
> each stage (so bytes will flow in/out as soon as they are available).
>
>
>
> The obvious solution would be to split the data into very small buffers,
> but since each segment would have to be sent completely to the same
> processor node (and not split between several nodes), doing such
> micro-batching would be a bad idea as it would spread a single segment’s
> buffers between multiple nodes.
>
>
>
> Is there any way to accomplish this with Flink? Or is Flink the wrong
> platform for that type of processing?
>
>
>
> Any help would be greatly appreciated!
>
>
>
> Thanks,
>
>
>
> Tal
>

parallelism parameter and output relation

[Serkan Taş]

I am working on this example 
http://www.itshared.org/2015/03/naive-bayes-on-apache-flink.html to learn get 
some more experience on platform.

Question is ;

By default, the output of process is double file (named 1 and 2) located in 
created folder. If i set parallelism to 1, FileNotFound exception is thrown.

I was expecting to get a single file instead, am i right ?


Serkan Taş
Mobil : +90 532 250 07 71
Likya Bilgi Teknolojileri
ve İletişim Hiz. Ltd. Şti.
www.likyateknoloji.com 
 
--
Bu elektronik posta ve onunla iletilen bütün dosyalar gizlidir. Sadece yukarıda 
isimleri belirtilen kişiler arasında özel haberleşme amacını taşımaktadır. Size 
yanlışlıkla ulaşmışsa bu elektonik postanın içeriğini açıklamanız, 
kopyalamanız, yönlendirmeniz ve kullanmanız kesinlikle yasaktır. Lütfen mesajı 
geri gönderiniz ve sisteminizden siliniz. Likya Bilgi Teknolojileri ve İletişim 
Hiz. Ltd. Şti. bu mesajın içeriği ile ilgili olarak hiç bir hukuksal 
sorumluluğu kabul etmez.
 
This electronic mail and any files transmitted with it are intended for the 
private use of  the persons named above. If you received this message in error, 
forwarding, copying or use of any of the information is strictly prohibited. 
Please immediately notify the sender and delete it from your system. Likya 
Bilgi Teknolojileri ve İletişim Hiz. Ltd. Şti. does not accept legal 
responsibility for the contents of this message.
--








P
Bu e-postayı yazdırmadan önce, çevreye olan sorumluluğunuzu tekrar düşünün.
Please consider your environmental responsibility before printing this e-mail.
 

Re: Frequent exceptions killing streaming job

[Robert Metzger]

Hey Nick,

I had a discussion with Stephan Ewen on how we could resolve the issue.
I filed a JIRA with our suggested approach:
https://issues.apache.org/jira/browse/FLINK-3264

By handling this directly in the KafkaConsumer, we would avoid fetching
data we can not handle anyways (discarding in the deserialization schema
would be more inefficient).

Let us know what you think about our suggested approach.

Sadly, it seems that the Kafka 0.9 consumer API does not yet support
requesting the latest offset of a TopicPartition. I'll ask about this on
their ML.




On Sun, Jan 17, 2016 at 8:28 PM, Nick Dimiduk  wrote:

> On Sunday, January 17, 2016, Stephan Ewen  wrote:
>
>> I agree, real time streams should never go down.
>>
>
>  Glad to hear that :)
>
>
>> [snip] Both should be supported.
>>
>
> Agreed.
>
>
>> Since we interpret streaming very broadly (also including analysis of
>> historic streams or timely data), the "backpressure/catch-up" mode seemed
>> natural as the first one to implement.
>>
>
> Indeed, this is what my job is doing. I have set it to, lacking a valid
> offset, start from the beginning. I have to presume that in my case the
> stream data is expiring faster than my consumers can keep up. However I
> haven't investigated proper monitoring yet.
>
>
>> The "load shedding" variant can probably even be realized in the Kafka
>> consumer, without complex modifications to the core Flink runtime itself.
>>
>
> I agree here as well. Indeed, this exception is being thrown from the
> consumer, not the runtime.
>
>
>
>> On Sun, Jan 17, 2016 at 12:42 AM, Nick Dimiduk 
>> wrote:
>>
>>> This goes back to the idea that streaming applications should never go
>>> down. I'd much rather consume at max capacity and knowingly drop some
>>> portion of the incoming pipe than have the streaming job crash. Of course,
>>> once the job itself is robust, I still need the runtime to be robust --
>>> YARN vs (potential) Mesos vs standalone cluster will be my next
>>> consideration.
>>>
>>> I can share some details about my setup, but not at this time; in part
>>> because I don't have my metrics available at the moment and in part because
>>> this is a public, archived list.
>>>
>>> On Sat, Jan 16, 2016 at 8:23 AM, Stephan Ewen  wrote:
>>>
 @Robert: Is it possible to add a "fallback" strategy to the consumer?
 Something like "if offsets cannot be found, use latest"?

 I would make this an optional feature to activate. I would think it is
 quite surprising to users if records start being skipped in certain
 situations. But I can see that this would be desirable sometimes.

 More control over skipping the records could be something to implement
 in an extended version of the Kafka Consumer. A user could define a policy
 that, in case consumer falls behind producer more than X (offsets), it
 starts requesting the latest offsets (rather than the following), thereby
 skipping a bunch of records.



 On Sat, Jan 16, 2016 at 3:14 PM, Robert Metzger 
 wrote:

> Hi Nick,
>
> I'm sorry you ran into the issue. Is it possible that Flink's Kafka
> consumer falls back in the topic so far that the offsets it's requesting
> are invalid?
>
> For that, the retention time of Kafka has to be pretty short.
>
> Skipping records under load is something currently not supported by
> Flink itself. The only idea I had for handling this would be to give the
> DeserializationSchema a call back to request the latest offset from Kafka
> to determine the lag. With that, the schema could determine a "dropping
> rate" to catch up.
> What would you as an application developer expect to handle the
> situation?
>
>
> Just out of curiosity: What's the throughput you have on the Kafka
> topic?
>
>
> On Fri, Jan 15, 2016 at 10:13 PM, Nick Dimiduk 
> wrote:
>
>> Hi folks,
>>
>> I have a streaming job that consumes from of a kafka topic. The topic
>> is pretty active so the local-mode single worker is obviously not able to
>> keep up with the fire-hose. I expect the job to skip records and continue
>> on. However, I'm getting an exception from the LegacyFetcher which kills
>> the job. This is very much *not* what I want. Any thoughts? The only 
>> thing
>> I find when I search for this error message is a link back to FLINK-2656.
>> I'm running roughly 0.10-release/HEAD.
>>
>> Thanks a lot,
>> Nick
>>
>> java.lang.Exception: Found invalid offsets more than once in
>> partitions [FetchPartition {partition=X, offset=Y}] Exceptions:
>> at
>> org.apache.flink.streaming.connectors.kafka.internals.LegacyFetcher.run(LegacyFetcher.java:242)
>> at
>> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer.run(FlinkKafkaConsumer.java:399)
>> at
>> org.apache.flink.streaming.api.operators.Strea

Re: Unexpected out of bounds error in UnilateralSortMerger

[Stephan Ewen]

Hi!

Does this error occur in 0.10 or im 1.0-SNAPSHOT?

It is probably an incorrectly configured Kryo instance (not a problem of
the sorter).
What is strange is that it occurs in the "MapReferenceResolver" - there
should be no reference resolution during serialization / deserialization.

Can you try what happens when you explicitly register the type SparseVector
at the ExecutionEnvironment?

Stephan


On Wed, Jan 20, 2016 at 11:24 AM, Theodore Vasiloudis <
theodoros.vasilou...@gmail.com> wrote:

> Hello all,
>
> I'm trying to run a job using FlinkML and I'm confused about the source of
> an error.
>
> The job reads a libSVM formatted file and trains an SVM classifier on it.
>
> I've tried this with small datasets and everything works out fine.
>
> When trying to run the same job on a large dataset (~11GB uncompressed)
> however, I get the following error:
>
>
>> java.lang.RuntimeException: Error obtaining the sorted input: Thread
>> 'SortMerger spilling thread' terminated due to an exception:
>> java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
>> Serialization trace:
>> indices (org.apache.flink.ml.math.SparseVector)
>> at
>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.getIterator(UnilateralSortMerger.java:619)
>> at
>> org.apache.flink.runtime.operators.BatchTask.getInput(BatchTask.java:1089)
>> at
>> org.apache.flink.runtime.operators.NoOpDriver.run(NoOpDriver.java:78)
>> at
>> org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:489)
>> at
>> org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:354)
>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
>> at java.lang.Thread.run(Thread.java:745)
>> Caused by: java.io.IOException: Thread 'SortMerger spilling thread'
>> terminated due to an exception: java.lang.IndexOutOfBoundsException: Index:
>> 14, Size: 2
>> Serialization trace:
>> indices (org.apache.flink.ml.math.SparseVector)
>> at
>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:800)
>> Caused by: com.esotericsoftware.kryo.KryoException:
>> java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
>> Serialization trace:
>> indices (org.apache.flink.ml.math.SparseVector)
>> at
>> com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125)
>> at
>> com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
>> at
>> com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
>> at
>> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:222)
>> at
>> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:236)
>> at
>> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:246)
>> at
>> org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase.copy(TupleSerializerBase.java:73)
>> at
>> org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase.copy(TupleSerializerBase.java:73)
>> at
>> org.apache.flink.runtime.operators.sort.NormalizedKeySorter.writeToOutput(NormalizedKeySorter.java:499)
>> at
>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger$SpillingThread.go(UnilateralSortMerger.java:1344)
>> at
>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:796)
>> Caused by: java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
>> at java.util.ArrayList.rangeCheck(ArrayList.java:653)
>> at java.util.ArrayList.set(ArrayList.java:444)
>> at
>> com.esotericsoftware.kryo.util.MapReferenceResolver.setReadObject(MapReferenceResolver.java:38)
>> at com.esotericsoftware.kryo.Kryo.reference(Kryo.java:823)
>> at com.esotericsoftware.kryo.Kryo.readObjectOrNull(Kryo.java:731)
>> at
>> com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:113)
>> ... 10 more
>
>
>
> Any idea what might be causing this? I'm running the job in local mode, 1
> TM with 8 slots and ~32GB heap size.
>
> All the vectors created by the libSVM loader have the correct size.
>

Unexpected out of bounds error in UnilateralSortMerger

[Theodore Vasiloudis]

Hello all,

I'm trying to run a job using FlinkML and I'm confused about the source of
an error.

The job reads a libSVM formatted file and trains an SVM classifier on it.

I've tried this with small datasets and everything works out fine.

When trying to run the same job on a large dataset (~11GB uncompressed)
however, I get the following error:


> java.lang.RuntimeException: Error obtaining the sorted input: Thread
> 'SortMerger spilling thread' terminated due to an exception:
> java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
> Serialization trace:
> indices (org.apache.flink.ml.math.SparseVector)
> at
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.getIterator(UnilateralSortMerger.java:619)
> at
> org.apache.flink.runtime.operators.BatchTask.getInput(BatchTask.java:1089)
> at
> org.apache.flink.runtime.operators.NoOpDriver.run(NoOpDriver.java:78)
> at
> org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:489)
> at
> org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:354)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
> at java.lang.Thread.run(Thread.java:745)
> Caused by: java.io.IOException: Thread 'SortMerger spilling thread'
> terminated due to an exception: java.lang.IndexOutOfBoundsException: Index:
> 14, Size: 2
> Serialization trace:
> indices (org.apache.flink.ml.math.SparseVector)
> at
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:800)
> Caused by: com.esotericsoftware.kryo.KryoException:
> java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
> Serialization trace:
> indices (org.apache.flink.ml.math.SparseVector)
> at
> com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125)
> at
> com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
> at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
> at
> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:222)
> at
> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:236)
> at
> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:246)
> at
> org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase.copy(TupleSerializerBase.java:73)
> at
> org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase.copy(TupleSerializerBase.java:73)
> at
> org.apache.flink.runtime.operators.sort.NormalizedKeySorter.writeToOutput(NormalizedKeySorter.java:499)
> at
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger$SpillingThread.go(UnilateralSortMerger.java:1344)
> at
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:796)
> Caused by: java.lang.IndexOutOfBoundsException: Index: 14, Size: 2
> at java.util.ArrayList.rangeCheck(ArrayList.java:653)
> at java.util.ArrayList.set(ArrayList.java:444)
> at
> com.esotericsoftware.kryo.util.MapReferenceResolver.setReadObject(MapReferenceResolver.java:38)
> at com.esotericsoftware.kryo.Kryo.reference(Kryo.java:823)
> at com.esotericsoftware.kryo.Kryo.readObjectOrNull(Kryo.java:731)
> at
> com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:113)
> ... 10 more



Any idea what might be causing this? I'm running the job in local mode, 1
TM with 8 slots and ~32GB heap size.

All the vectors created by the libSVM loader have the correct size.

Actual byte-streams in multiple-node pipelines

[Tal Maoz]

Hey,



I’m a new user to Flink and I’m trying to figure out if I can build a
pipeline I’m working on using Flink.

I have a data source that sends out a continues data stream at a bandwidth
of anywhere between 45MB/s to 600MB/s (yes, that’s MiB/s, not Mib/s, and
NOT a series of individual messages but an actual continues stream of data
where some data may depend on previous or future data to be fully
deciphered).

I need to be able to pass the data through several processing stages (that
manipulate the data but still produce the same order of magnitude output at
each stage) and I need processing to be done with low-latency.

The data itself CAN be segmented but the segments will be some HUGE (~100MB
– 250MB) and I would like to be able to stream data in and out of the
processors ASAP instead of waiting for full segments to be complete at each
stage (so bytes will flow in/out as soon as they are available).



The obvious solution would be to split the data into very small buffers,
but since each segment would have to be sent completely to the same
processor node (and not split between several nodes), doing such
micro-batching would be a bad idea as it would spread a single segment’s
buffers between multiple nodes.



Is there any way to accomplish this with Flink? Or is Flink the wrong
platform for that type of processing?



Any help would be greatly appreciated!



Thanks,



Tal

Re: flink 1.0-SNAPSHOT scala 2.11 compilation error

[Robert Metzger]

Hey David,

the issue should be resolved now. Please let me know if its still an issue
for you.

Regards,
Robert


On Fri, Jan 15, 2016 at 4:02 PM, David Kim 
wrote:

> Thanks Till! I'll keep an eye out on the JIRA issue. Many thanks for the
> prompt reply.
>
> Cheers,
> David
>
> On Fri, Jan 15, 2016 at 4:16 AM, Till Rohrmann 
> wrote:
>
>> Hi David,
>>
>> this is definitely an error on our side which might be caused by the
>> latest changes to the project structure (removing flink-staging directory).
>> I’ve filed a JIRA issue https://issues.apache.org/jira/browse/FLINK-3241.
>> It should be fixed soon.
>>
>> In the meantime it should work if you build yourself the Flink binaries
>> from the sources. First check them out, then call 
>> tools/change-scala-version.sh
>> 2.11 in the root directory and then mvn clean install -DskipTests
>> -Dmaven.javadoc.skip=true. These binaries should depend on the right
>> Scala version.
>>
>> Cheers,
>> Till
>> ​
>>
>> On Thu, Jan 14, 2016 at 11:25 PM, David Kim <
>> david@braintreepayments.com> wrote:
>>
>>> Hi,
>>>
>>> I have a scala project depending on flink scala_2.11 and am seeing a
>>> compilation error when using sbt.
>>>
>>> I'm using flink 1.0-SNAPSHOT and my build was working yesterday. I was
>>> wondering if maybe a recent change to flink could be the cause?
>>>
>>> Usually we see flink resolving the scala _2.11 counterparts for akka and
>>> scalatest:
>>>
>>> [info] Resolving com.typesafe.akka#akka-actor_2.11;2.3.7 ...
>>> [info] Resolving com.typesafe#config;1.2.1 ...
>>> [info] Resolving com.typesafe.akka#akka-remote_2.11;2.3.7 ...[info] 
>>> Resolving org.scalatest#scalatest_2.11;2.2.4 ...
>>>
>>>
>>>
>>> but am seeing it pick up the _2.10 counterparts:
>>>
>>> [info] Resolving com.typesafe.akka#akka-actor_2.10;2.3.7 ...[info] 
>>> Resolving com.typesafe.akka#akka-actor_2.10;2.3.7 ...
>>> [info] Resolving com.typesafe#config;1.2.1 ...[info] Resolving 
>>> org.scalatest#scalatest_2.10;2.2.2 ...
>>>
>>>
>>>
>>> This ultimately gives us the following compilation error:
>>>
>>> [error]org.scalatest:scalatest _2.10, _2.11
>>> java.lang.RuntimeException: Conflicting cross-version suffixes in: 
>>> org.scalatest:scalatest
>>> at scala.sys.package$.error(package.scala:27)
>>> at sbt.ConflictWarning$.processCrossVersioned(ConflictWarning.scala:46)
>>> at sbt.ConflictWarning$.apply(ConflictWarning.scala:32)
>>>
>>>
>>>
>>> Maybe the latest 1.0-SNAPSHOT build erroneously built with scala 2.10
>>> for the 2.11 profile? Any guidance appreciated!
>>>
>>> Thanks,
>>> David
>>>
>>>
>>
>
>
> --
> Note: this information is confidential. It is prohibited to share, post
> online or otherwise publicize without Braintree's prior written consent.
>

Re: Compile fails with scala 2.11.4

[Robert Metzger]

Hi,

in the latest master, the "tools/change-scala-version.sh" should be fixed.
Also, the 1.0-SNAPSHOT version deployed to the snapshot repository should
be good again.

@Ritesh: The commands were correct. I'm not sure if Flink builds with Scala
2.11.4, the default 2.11 version we are using is 2.11.7.



On Tue, Jan 19, 2016 at 7:41 AM, Prez Cannady 
wrote:

> Assuming you haven’t already migrated back to 2.10, you might try this;
> $ git checkout release-0.10
> $ tools/change-scala-version 2.11
> $ mvn clean install -DskipTests=true -Dmaven.javadoc.skip=true
> -Dscala.version=2.11.4 -Dscala.binary.version=2.11
>
> Then try building your project.  Building under these conditions resolved
> my issue which emitted the same error.
>
> Prez Cannady
> p: 617 500 3378
> e: revp...@opencorrelate.org
> GH: https://github.com/opencorrelate
> LI: https://www.linkedin.com/in/revprez
>
>
>
>
>
>
>
>
>
> On Jan 18, 2016, at 10:20 AM, Ritesh Kumar Singh <
> riteshoneinamill...@gmail.com> wrote:
>
> Thanks for the replies.
>
> @Chiwan, I am switching back to scala_2.10.4 for the time being. I was
> using scala_2.11.4 as this is the version I've compiled spark with. But
> anyways, I can wait for the bug to be resolved.
>
> @Robert, the commands were as follows:
> $tools/change-scala-version.sh 2.11
> $mvn clean install -DskipTests -Dscala.version=2.11.4
>
> I hope I'm doing it right ?
>
> Thanks,
>
> *Ritesh Kumar Singh,*
> *https://riteshtoday.wordpress.com/* 
>
> On Mon, Jan 18, 2016 at 12:03 PM, Robert Metzger 
> wrote:
>
>> How did start the Flink for Scala 2.11 compilation ?
>>
>> On Mon, Jan 18, 2016 at 11:41 AM, Chiwan Park 
>> wrote:
>>
>>> Hi Ritesh,
>>>
>>> This problem seems already reported [1]. Flink community is
>>> investigating this issue. I think that if you don’t need Scala 2.11, use
>>> Scala 2.10 until the issue is solved.
>>>
>>> [1]:
>>> http://mail-archives.apache.org/mod_mbox/flink-user/201601.mbox/%3CCAB6CeiZ_2snN-piXzd3gHnyQePu_PA0Ro7qXUF8%3DVTxoyL0YyA%40mail.gmail.com%3E
>>>
>>> > On Jan 18, 2016, at 7:24 PM, Ritesh Kumar Singh <
>>> riteshoneinamill...@gmail.com> wrote:
>>> >
>>> > [ERROR]
>>> /home/flink/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala:703:
>>> error: can't expand macros compiled by previous versions of Scala
>>> > [ERROR]   assert(cachedGraph2.isArchived)
>>> > [ERROR]   ^
>>> > [ERROR] one error found
>>> > [INFO]
>>> 
>>> > [INFO] Reactor Summary:
>>> > [INFO]
>>> > [INFO] flink .. SUCCESS [
>>> 24.820 s]
>>> > [INFO] flink-annotations .. SUCCESS [
>>> 2.755 s]
>>> > [INFO] flink-shaded-hadoop  SUCCESS [
>>> 0.208 s]
>>> > [INFO] flink-shaded-hadoop2 ... SUCCESS [
>>> 15.627 s]
>>> > [INFO] flink-shaded-include-yarn-tests  SUCCESS [
>>> 17.076 s]
>>> > [INFO] flink-shaded-curator ... SUCCESS [
>>> 0.200 s]
>>> > [INFO] flink-shaded-curator-recipes ... SUCCESS [
>>> 2.751 s]
>>> > [INFO] flink-shaded-curator-test .. SUCCESS [
>>> 0.355 s]
>>> > [INFO] flink-core . SUCCESS [
>>> 33.052 s]
>>> > [INFO] flink-java . SUCCESS [
>>> 10.224 s]
>>> > [INFO] flink-runtime .. FAILURE
>>> [01:23 min]
>>> > [INFO] flink-optimizer  SKIPPED
>>> >
>>> >
>>> > Any workaround for scala_2.11.4 or do I have to switch back to
>>> scala_2.10.4 ?
>>> >
>>> > Thanks,
>>> > Ritesh Kumar Singh,
>>> > https://riteshtoday.wordpress.com/
>>> >
>>>
>>> Regards,
>>> Chiwan Park
>>>
>>>
>>
>
>