Just tried the following and it worked:
public static void main(String[] args) throws IOException {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
final DataStreamSource<Integer> source = env.fromElements(1, 2, 3, 4);
source.print();
final Iterator<Integer> iter = DataStreamUtils.collect(source);
while (iter.hasNext()) {
System.out.println(iter.next());
}
}
It prints:
1
2
3
4
2> 2
1> 1
4> 4
3> 3
However, the collect util needs some improvements. It assumes that the
machine running the code is reachable on a random port by the Flink
cluster. If you have any firewalls, then this might not work.
Cheers,
Max
On Tue, Jul 19, 2016 at 10:13 PM, subash basnet <yasub...@gmail.com> wrote:
> Hello Till,
>
> Yup I can see the log output in my console, but there is no information
> there regarding if there is any error in conversion. Just normal warn and
> info as below:
> 22:09:16,676 WARN org.apache.flink.streaming.runtime.tasks.StreamTask
> - No state backend has been specified, using default state backend
> (Memory / JobManager)
> 22:09:16,676 INFO org.apache.flink.streaming.runtime.tasks.StreamTask
> - State backend is set to heap memory (checkpoint to jobmanager)
>
> The above message is always there when I run my project. It would be great
> if someone could check why the collection of datastream via DataStreamUtils
> is giving empty result.
>
> Best Regards,
> Subash Basnet
>
> On Tue, Jul 19, 2016 at 4:52 PM, Till Rohrmann <trohrm...@apache.org>
> wrote:
>
>> It depends if you have a log4j.properties file specified in your
>> classpath. If you see log output on the console, then it should also print
>> errors there.
>>
>> Cheers,
>> Till
>>
>> On Tue, Jul 19, 2016 at 3:08 PM, subash basnet <yasub...@gmail.com>
>> wrote:
>>
>>> Hello Till,
>>>
>>> Shouldn't it write something in the eclipse console if there is any
>>> error or warning. But nothing about error is printed on the console. And I
>>> checked the flink project folder: flink-core, flink streaming as such but
>>> couldn't find where the log is written when run via eclipse.
>>>
>>> Best Regards,
>>> Subash Basnet
>>>
>>> On Tue, Jul 19, 2016 at 2:49 PM, Till Rohrmann <trohrm...@apache.org>
>>> wrote:
>>>
>>>> Have you checked your logs whether they contain some problems? In
>>>> general it is not recommended collecting the streaming result back to your
>>>> client. It might also be a problem with `DataStreamUtils.collect`.
>>>>
>>>> Cheers,
>>>> Till
>>>>
>>>> On Tue, Jul 19, 2016 at 2:42 PM, subash basnet <yasub...@gmail.com>
>>>> wrote:
>>>>
>>>>> Hello all,
>>>>>
>>>>> I tried to check if it works for tuple but same problem, the
>>>>> collection still shows blank result. I took the id of centroid tuple and
>>>>> printed it, but the collection displays empty.
>>>>>
>>>>> DataStream<Centroid> centroids = newCentroidDataStream.map(new
>>>>> TupleCentroidConverter());
>>>>> DataStream<Tuple1<String>> centroidId = centroids.map(new
>>>>> TestMethod());
>>>>> centroidId.print();
>>>>> Iterator<Tuple1<String>> iter = DataStreamUtils.collect(centroidId);
>>>>> Collection<Tuple1<String>> testCentroids = Lists.newArrayList(iter);
>>>>> for (Tuple1<String> c : testCentroids) {
>>>>> System.out.println(c);
>>>>> }
>>>>> Output: (Mon Jul 18 17:36:03 CEST 2016) (Mon Jul 18 17:43:58 CEST
>>>>> 2016) (Mon Jul 18 17:42:59 CEST 2016) (Mon Jul 18 17:34:01 CEST 2016) (Mon
>>>>> Jul 18 17:52:00 CEST 2016) (Mon Jul 18 17:40:58 CEST 2016) for
>>>>> centroidId.print(), but no output for System.out.println(c); Best Regards,
>>>>> Subash Basnet
>>>>>
>>>>> On Tue, Jul 19, 2016 at 10:48 AM, subash basnet <yasub...@gmail.com>
>>>>> wrote:
>>>>>
>>>>>> Hello all,
>>>>>>
>>>>>> I am trying to convert datastream to collection, but it's shows blank
>>>>>> result. There is a stream of data which can be viewed on the console on
>>>>>> print(), but the collection of the same stream shows empty after
>>>>>> conversion. Below is the code:
>>>>>>
>>>>>> DataStream<Centroid> centroids = newCentroidDataStream.map(new
>>>>>> TupleCentroidConverter());
>>>>>> centroids.print();
>>>>>> Iterator<Centroid> iter = DataStreamUtils.collect(centroids);
>>>>>> Collection<Centroid> testCentroids = Lists.newArrayList(iter);
>>>>>> for(Centroid c: testCentroids){
>>>>>> System.out.println(c);
>>>>>> }
>>>>>>
>>>>>> The above *centroids.print()* gives the following output in console:
>>>>>>
>>>>>> Mon Jul 18 21:29:01 CEST 2016 119.3701 119.4 119.3701 119.38 27400.0
>>>>>> Mon Jul 18 21:23:00 CEST 2016 119.3463 119.37 119.315 119.37 48200.0
>>>>>> Mon Jul 18 21:27:59 CEST 2016 119.3401 119.3401 119.26 119.265
>>>>>> 50300.0
>>>>>> Mon Jul 18 21:36:00 CEST 2016 119.48 119.505 119.47 119.4741 37400.0
>>>>>> Mon Jul 18 21:33:00 CEST 2016 119.535 119.54 119.445 119.455 152900.0
>>>>>>
>>>>>> But the next *System.out.println(c) *within the for loop prints
>>>>>> nothing. What could be the problem.
>>>>>>
>>>>>> My maven has following configuration for dataStreamUtils:
>>>>>> <dependency>
>>>>>> <groupId>org.apache.flink</groupId>
>>>>>> <artifactId>flink-streaming-contrib_2.10</artifactId>
>>>>>> <version>${flink.version}</version>
>>>>>> </dependency>
>>>>>>
>>>>>>
>>>>>> Best Regards,
>>>>>> Subash Basnet
>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>>
>
