how to convert datastream to collection

[subashbasnet]

Hello all,

Suppose I have the datastream as:
DataStream<Tuple2<String,Point>> newCentroids;

How to get collection of newCentroids  to be able to loop as below:
 private Collection<Tuple2<String,Point>> centroids;

 for (Centroid cent : centroids) {

  }



Best Regards,
Subash Basnet

[Suneel Marthi]

DataStream<Tuple2<String,Point>> newCentroids = new DataStream<>.....()

Iterator<Tuple2<String, Point>> iter = DataStreamUtils.collect(newCentroids);

List<Tuple2<>> list = Lists.newArrayList(iter);

On Tue, May 3, 2016 at 10:26 AM, subash basnet <[hidden email]> wrote:

Hello all,

Suppose I have the datastream as:
DataStream<Tuple2<String,Point>> newCentroids;

How to get collection of newCentroids  to be able to loop as below:
 private Collection<Tuple2<String,Point>> centroids;

 for (Centroid cent : centroids) {

  }



Best Regards,
Subash Basnet

[Aljoscha Krettek]

Hi,

please keep in mind that we're dealing with streams. The Iterator might never finish.

Cheers,

Aljoscha

On Tue, 3 May 2016 at 16:35 Suneel Marthi <[hidden email]> wrote:

DataStream<Tuple2<String,Point>> newCentroids = new DataStream<>.....()

Iterator<Tuple2<String, Point>> iter = DataStreamUtils.collect(newCentroids);

List<Tuple2<>> list = Lists.newArrayList(iter);

On Tue, May 3, 2016 at 10:26 AM, subash basnet <[hidden email]> wrote:

Hello all,

Suppose I have the datastream as:
DataStream<Tuple2<String,Point>> newCentroids;

How to get collection of newCentroids  to be able to loop as below:
 private Collection<Tuple2<String,Point>> centroids;

 for (Centroid cent : centroids) {

  }



Best Regards,
Subash Basnet

[Srikanth]

Why do you want collect and iterate? Why not iterate on the DataStream itself?

May be I didn't understand your use case completely.

Srikanth

On Tue, May 3, 2016 at 10:55 AM, Aljoscha Krettek <[hidden email]> wrote:

Hi,

please keep in mind that we're dealing with streams. The Iterator might never finish.

Cheers,

Aljoscha

On Tue, 3 May 2016 at 16:35 Suneel Marthi <[hidden email]> wrote:

DataStream<Tuple2<String,Point>> newCentroids = new DataStream<>.....()

Iterator<Tuple2<String, Point>> iter = DataStreamUtils.collect(newCentroids);

List<Tuple2<>> list = Lists.newArrayList(iter);

On Tue, May 3, 2016 at 10:26 AM, subash basnet <[hidden email]> wrote:

Hello all,

Suppose I have the datastream as:
DataStream<Tuple2<String,Point>> newCentroids;

How to get collection of newCentroids  to be able to loop as below:
 private Collection<Tuple2<String,Point>> centroids;

 for (Centroid cent : centroids) {

  }



Best Regards,
Subash Basnet

[subashbasnet]

Hello there,

Thank you! But I couldn't find DataStreamUtils. Where is it located?
I am using <flink.version>1.0.0</flink.version>, 
The given, 

org.apache.flink.contrib.streaming.DataStreamUtils doesn't work as there is no contrib package within org.apache.flink. 


Best Regards,
Subash Basnet

On Tue, May 3, 2016 at 4:35 PM, Suneel Marthi <[hidden email]> wrote:

This message is eligible for Automatic Cleanup! ([hidden email]) Add cleanup rule | More info

DataStream<Tuple2<String,Point>> newCentroids = new DataStream<>.....()

Iterator<Tuple2<String, Point>> iter = DataStreamUtils.collect(newCentroids);

List<Tuple2<>> list = Lists.newArrayList(iter);

On Tue, May 3, 2016 at 10:26 AM, subash basnet <[hidden email]> wrote:

Hello all,

Suppose I have the datastream as:
DataStream<Tuple2<String,Point>> newCentroids;

How to get collection of newCentroids  to be able to loop as below:
 private Collection<Tuple2<String,Point>> centroids;

 for (Centroid cent : centroids) {

  }



Best Regards,
Subash Basnet

[stefanobaghino]

I think you have to explicitly import contrib packages as a dependency in Maven/SBT.

In particular, in your case, it should be something like:

<dependency>

<groupId>org.apache.flink</groupId>

<artifactId>flink-streaming-contrib_${scala.version}</artifactId>

<version>${project.version}</version>

</dependency>

for Maven.

On Wed, May 4, 2016 at 12:00 PM, subash basnet <[hidden email]> wrote:

Hello there,

Thank you! But I couldn't find DataStreamUtils. Where is it located?
I am using <flink.version>1.0.0</flink.version>, 
The given, 

org.apache.flink.contrib.streaming.DataStreamUtils doesn't work as there is no contrib package within org.apache.flink. 


Best Regards,
Subash Basnet

On Tue, May 3, 2016 at 4:35 PM, Suneel Marthi <[hidden email]> wrote:

This message is eligible for Automatic Cleanup! ([hidden email]) Add cleanup rule | More info

DataStream<Tuple2<String,Point>> newCentroids = new DataStream<>.....()

Iterator<Tuple2<String, Point>> iter = DataStreamUtils.collect(newCentroids);

List<Tuple2<>> list = Lists.newArrayList(iter);

On Tue, May 3, 2016 at 10:26 AM, subash basnet <[hidden email]> wrote:

Hello all,

Suppose I have the datastream as:
DataStream<Tuple2<String,Point>> newCentroids;

How to get collection of newCentroids  to be able to loop as below:
 private Collection<Tuple2<String,Point>> centroids;

 for (Centroid cent : centroids) {

  }



Best Regards,
Subash Basnet

--

[subashbasnet]

Hello Stefano,

When I did as you said, it shows: Missing artifact org.apache.flink:flink-streaming-contrib_1.0.0:jar:1.0.0 error.
My pom is:

<properties>

<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

<flink.version>1.0.0</flink.version>

</properties>

<dependency>

<groupId>org.apache.flink</groupId>

<artifactId>flink-streaming-contrib_${flink.version}</artifactId>

<version>${flink.version}</version>

</dependency>

So I instead used the below for contrib dependency as given in maven repository:

<dependency>

<groupId>org.apache.flink</groupId>

<artifactId>flink-streaming-contrib</artifactId>

<version>0.10.2</version>

</dependency>


Best Regards,
Subash Basnet

On Wed, May 4, 2016 at 12:12 PM, Stefano Baghino <[hidden email]> wrote:

I think you have to explicitly import contrib packages as a dependency in Maven/SBT.

In particular, in your case, it should be something like:

<dependency>

<groupId>org.apache.flink</groupId>

<artifactId>flink-streaming-contrib_${scala.version}</artifactId>

<version>${project.version}</version>

</dependency>

for Maven.

On Wed, May 4, 2016 at 12:00 PM, subash basnet <[hidden email]> wrote:

Hello there,

Thank you! But I couldn't find DataStreamUtils. Where is it located?
I am using <flink.version>1.0.0</flink.version>, 
The given, 

org.apache.flink.contrib.streaming.DataStreamUtils doesn't work as there is no contrib package within org.apache.flink. 


Best Regards,
Subash Basnet

On Tue, May 3, 2016 at 4:35 PM, Suneel Marthi <[hidden email]> wrote:

This message is eligible for Automatic Cleanup! ([hidden email]) Add cleanup rule | More info

DataStream<Tuple2<String,Point>> newCentroids = new DataStream<>.....()

Iterator<Tuple2<String, Point>> iter = DataStreamUtils.collect(newCentroids);

List<Tuple2<>> list = Lists.newArrayList(iter);

On Tue, May 3, 2016 at 10:26 AM, subash basnet <[hidden email]> wrote:

Hello all,

Suppose I have the datastream as:
DataStream<Tuple2<String,Point>> newCentroids;

How to get collection of newCentroids  to be able to loop as below:
 private Collection<Tuple2<String,Point>> centroids;

 for (Centroid cent : centroids) {

  }



Best Regards,
Subash Basnet

--

BR,

Stefano Baghino

Software Engineer @ Radicalbit

[Aljoscha Krettek]

<dependency>

    <groupId>org.apache.flink</groupId>

    <artifactId>flink-streaming-contrib_2.10</artifactId>

    <version>1.0.0</version>

</dependency>

On Wed, 4 May 2016 at 12:44 subash basnet <[hidden email]> wrote:

Hello Stefano,

When I did as you said, it shows: Missing artifact org.apache.flink:flink-streaming-contrib_1.0.0:jar:1.0.0 error.
My pom is:

<properties>

<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

<flink.version>1.0.0</flink.version>

</properties>

<dependency>

<groupId>org.apache.flink</groupId>

<artifactId>flink-streaming-contrib_${flink.version}</artifactId>

<version>${flink.version}</version>

</dependency>

So I instead used the below for contrib dependency as given in maven repository:

<dependency>

<groupId>org.apache.flink</groupId>

<artifactId>flink-streaming-contrib</artifactId>

<version>0.10.2</version>

</dependency>


Best Regards,
Subash Basnet

On Wed, May 4, 2016 at 12:12 PM, Stefano Baghino <[hidden email]> wrote:

I think you have to explicitly import contrib packages as a dependency in Maven/SBT.

In particular, in your case, it should be something like:

<dependency>

<groupId>org.apache.flink</groupId>

<artifactId>flink-streaming-contrib_${scala.version}</artifactId>

<version>${project.version}</version>

</dependency>

for Maven.

On Wed, May 4, 2016 at 12:00 PM, subash basnet <[hidden email]> wrote:

Hello there,

Thank you! But I couldn't find DataStreamUtils. Where is it located?
I am using <flink.version>1.0.0</flink.version>, 
The given, 

org.apache.flink.contrib.streaming.DataStreamUtils doesn't work as there is no contrib package within org.apache.flink. 


Best Regards,
Subash Basnet

On Tue, May 3, 2016 at 4:35 PM, Suneel Marthi <[hidden email]> wrote:

This message is eligible for Automatic Cleanup! ([hidden email]) Add cleanup rule | More info

DataStream<Tuple2<String,Point>> newCentroids = new DataStream<>.....()

Iterator<Tuple2<String, Point>> iter = DataStreamUtils.collect(newCentroids);

List<Tuple2<>> list = Lists.newArrayList(iter);

On Tue, May 3, 2016 at 10:26 AM, subash basnet <[hidden email]> wrote:

Hello all,

Suppose I have the datastream as:
DataStream<Tuple2<String,Point>> newCentroids;

How to get collection of newCentroids  to be able to loop as below:
 private Collection<Tuple2<String,Point>> centroids;

 for (Centroid cent : centroids) {

  }



Best Regards,
Subash Basnet

--

BR,

Stefano Baghino

Software Engineer @ Radicalbit

[stefanobaghino]

I believe you may suffer from some undefined behaviour by using different versions for Flink and contrib libraries.

Please note that in your dependency declaration there's a typo: flink-streaming-contrib_${flink.version} should be flink-streaming-contrib_${scala.version}, with ${scala.version} being the one you use, either 2.10 or 2.11.

On Wed, May 4, 2016 at 12:43 PM, subash basnet <[hidden email]> wrote:

Hello Stefano,

When I did as you said, it shows: Missing artifact org.apache.flink:flink-streaming-contrib_1.0.0:jar:1.0.0 error.
My pom is:

<properties>

<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

<flink.version>1.0.0</flink.version>

</properties>

<dependency>

<groupId>org.apache.flink</groupId>

<artifactId>flink-streaming-contrib_${flink.version}</artifactId>

<version>${flink.version}</version>

</dependency>

So I instead used the below for contrib dependency as given in maven repository:

<dependency>

<groupId>org.apache.flink</groupId>

<artifactId>flink-streaming-contrib</artifactId>

<version>0.10.2</version>

</dependency>


Best Regards,
Subash Basnet

On Wed, May 4, 2016 at 12:12 PM, Stefano Baghino <[hidden email]> wrote:

I think you have to explicitly import contrib packages as a dependency in Maven/SBT.

In particular, in your case, it should be something like:

<dependency>

<groupId>org.apache.flink</groupId>

<artifactId>flink-streaming-contrib_${scala.version}</artifactId>

<version>${project.version}</version>

</dependency>

for Maven.

On Wed, May 4, 2016 at 12:00 PM, subash basnet <[hidden email]> wrote:

Hello there,

Thank you! But I couldn't find DataStreamUtils. Where is it located?
I am using <flink.version>1.0.0</flink.version>, 
The given, 

org.apache.flink.contrib.streaming.DataStreamUtils doesn't work as there is no contrib package within org.apache.flink. 


Best Regards,
Subash Basnet

On Tue, May 3, 2016 at 4:35 PM, Suneel Marthi <[hidden email]> wrote:

This message is eligible for Automatic Cleanup! ([hidden email]) Add cleanup rule | More info

DataStream<Tuple2<String,Point>> newCentroids = new DataStream<>.....()

Iterator<Tuple2<String, Point>> iter = DataStreamUtils.collect(newCentroids);

List<Tuple2<>> list = Lists.newArrayList(iter);

On Tue, May 3, 2016 at 10:26 AM, subash basnet <[hidden email]> wrote:

Hello all,

Suppose I have the datastream as:
DataStream<Tuple2<String,Point>> newCentroids;

How to get collection of newCentroids  to be able to loop as below:
 private Collection<Tuple2<String,Point>> centroids;

 for (Centroid cent : centroids) {

  }



Best Regards,
Subash Basnet

--

BR,

Stefano Baghino

Software Engineer @ Radicalbit

--