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
