Re: Count of Grouped DataSet
Hello all, This is how I have moved ahead with the implementation of finding count of a GroupedDataSet: *val k = envDefault .fromElements((1,1,2,"A"),(1,1,2,"B"),(2,1,3,"B"),(3,1,4,"C")) .groupBy(1,2) .reduceGroup(nextGroup => { val asList = nextGroup.toList (asList.head._2,asList.head._3,asList.size) }) k.print()* While this produces the expected output alright, I am not sure if this the ideal, idiomatic way to implement what I need. Could you please confirm? If there is a better way, I would like to be wiser of course. -- Nirmalya -- View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Count-of-Grouped-DataSet-tp6592p6594.html Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.
Re: Unable to write stream as csv
Hello there, If anyone could help me know why the below *result* DataStream get's written as text, but not as csv?. As it's in a tuple format I guess it should be the same for both text and csv. It shows no error just simply doesn't write to file when result is written as csv. DataStream> *result* = keyedEdits.timeWindow(Time.seconds(5)).fold(new Tuple2<>("", 0L), new FoldFunction>() { @Override public Tuple2 fold(Tuple2 acc, WikipediaEditEvent event) { acc.f0 = event.getUser(); acc.f1 += event.getByteDiff(); return acc; } }); *result.writeAsText(.); --> It is working. **result.writeAsCsv(.); ---> It is not working. * Best Regards, Subash Basnet On Wed, Apr 27, 2016 at 4:14 PM, subash basnet wrote: > Hello all, > > I am able to write the Wikipedia edit data to the kafka and as a text file > as per the given example of WikipediaAnalysis. But when I try to write it > as csv, the blank files initially created never gets filled with data. > Below is the code: > > DataStream> result = > keyedEdits.timeWindow(Time.seconds(5)).fold(new Tuple2<>("", 0L), > new FoldFunction>() { > @Override > public Tuple2 fold(Tuple2 acc, > WikipediaEditEvent event) { > acc.f0 = event.getUser(); > acc.f1 += event.getByteDiff(); > return acc; > } > }); > > *result.writeAsText("file:///home/softwares/flink-1.0.0/kmeans/streaming/result", > FileSystem.WriteMode.OVERWRITE); *-> works > > *result.writeAsCsv("file:///home/softwares/flink-1.0.0/kmeans/streaming/result", > FileSystem.WriteMode.OVERWRITE);* --> doesn't work > > Why is data getting written to file as text but not as csv? > > Best Regards, > Subash Basnet > >
Problem with writeAsText
I have a Dataset which contains only strings. But when I execute a writeAsText and supply a folder inside the string, it finishes with the following output but does not write any text files: org.apache.flink.api.java.operators.DataSink[String] = DataSink '' (TextOutputFormat (file://path/to/output) - UTF-8) -- Thank You Regards Punit Naik