You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Lea Helmers <le...@gmail.com> on 2015/11/01 13:52:47 UTC
Fwd: Problem applying a groupReduce function to a grouped data set
Hi!
When I try to apply a groupReduce function to a data set I get an error.
The data set is created like this:
DataSet<Tuple3<String, String, String>> actorsTemp =
env.readCsvFile("/home/lea/Documents/impro3_ws15/actors.tsv")
.fieldDelimiter("\t")
.includeFields("1110")
.types(String.class, String.class, String.class);
DataSet<Tuple3<String, String, String>> actresses =
env.readCsvFile("/home/lea/Documents/impro3_ws15/actresses.tsv")
.fieldDelimiter("\t")
.includeFields("1110")
.types(String.class, String.class, String.class);
DataSet<Tuple3<Float, Float, String>> ratings =
env.readCsvFile("/home/lea/Documents/impro3_ws15/ratings.tsv")
.fieldDelimiter("\t")
.includeFields("0111")
.types(Float.class, Float.class, String.class)
.filter(new NumberVotesFilter());
//merge actors and actresses
DataSet<Tuple3<String, String, String>> actors =
actorsTemp.union(actresses);
//create weighted rating
DataSet<Tuple2<String, Float>> weightedRatings =
ratings.map(new WeightedRatingCalculator());
THIS IS WHAT I'M TRYING IN THE MAIN METHOD:
actors.map(new JoinNames())
.join(weightedRatings)
.where(1).equalTo(0)
.projectFirst(0).projectSecond(1)
.groupBy(0)
.reduceGroup(new MeanRatingCalculator())
.first(10).print();
And here is the GroupReduce function I wrote:
public static class MeanRatingCalculator implements
GroupReduceFunction<Tuple2<String, Float>, Tuple3<String, Float, Integer>> {
public void reduce(Iterable<Tuple2<String, Float>> ratedActors,
Collector<Tuple3<String, Float, Integer>> out) throws Exception {
String name = null;
Float ratings = 0F;
int numberOfMovies = 0;
for (Tuple2<String, Float> a : ratedActors) {
//store the name
name = a.f0;
//update the sum of the ratings and number of movies
ratings += a.f1;
numberOfMovies++;
}
// emit name, average rating and number of films
out.collect(new Tuple3<String, Float, Integer>(name,
ratings/(float)numberOfMovies, numberOfMovies));
}
}
I get the following error message when I try to compile the code:
java: method reduceGroup in class
org.apache.flink.api.java.operators.UnsortedGrouping<T> cannot be applied
to given types;
required:
org.apache.flink.api.common.functions.GroupReduceFunction<org.apache.flink.api.java.tuple.Tuple,R>
found: de.tub.dima.TopActors.MeanRatingCalculator
reason: no instance(s) of type variable(s) R exist so that argument type
de.tub.dima.TopActors.MeanRatingCalculator conforms to formal parameter
type
org.apache.flink.api.common.functions.GroupReduceFunction<org.apache.flink.api.java.tuple.Tuple,R>
I can't figure out what the problem might be and would be very grateful for
any help!! I hope I have given all the necessary information. I'm using
Ubuntu 14.04 and IntelliJ Idea as IDE.
Thank you very much,
Lea
Re: Fwd: Problem applying a groupReduce function to a grouped data
set
Posted by Martin Junghanns <m....@mailbox.org>.
Hi,
just an idea: In the source code documentation, it states that
projectFirst and projectSecond lose type information, which could
explain why your group reduce expects <Tuple, R>.
I found an example [1] that calls .types() to define the returned types,
but this method is deprecated. What I would try is to replace the
projectFirst and projectSecond with a JoinFunction and output
Tuple2<String, Float> "manually" like so:
actors.map(new JoinNames())
.join(weightedRatings)
.where(1).equalTo(0)
.with(new JoinFunction<
TypeLeft, // output type of JoinNames()
Tuple2<String, Float>,
Tuple2<String, Float>> {
@Override
public Tuple2<String, Float> join(
TypeLeft left,
Tuple2<String, Float> right) throws Exception {
return new Tuple2<>(left.f0, right.f1);
}
})
.withForwardedFieldsFirst("f0")
.withForwardedFieldsSecond("f1")
.groupBy(0)
.reduceGroup(new MeanRatingCalculator())
.first(10)
.print();
Hope this helps.
Best,
Martin
[1]
https://ci.apache.org/projects/flink/flink-docs-master/apis/examples.html#relational-query
On 01.11.2015 13:52, Lea Helmers wrote:
> Hi!
>
> When I try to apply a groupReduce function to a data set I get an error.
>
> The data set is created like this:
>
> DataSet<Tuple3<String, String, String>> actorsTemp =
>
> env.readCsvFile("/home/lea/Documents/impro3_ws15/actors.tsv")
> .fieldDelimiter("\t")
> .includeFields("1110")
> .types(String.class, String.class, String.class);
>
> DataSet<Tuple3<String, String, String>> actresses =
>
> env.readCsvFile("/home/lea/Documents/impro3_ws15/actresses.tsv")
> .fieldDelimiter("\t")
> .includeFields("1110")
> .types(String.class, String.class, String.class);
>
> DataSet<Tuple3<Float, Float, String>> ratings =
>
> env.readCsvFile("/home/lea/Documents/impro3_ws15/ratings.tsv")
> .fieldDelimiter("\t")
> .includeFields("0111")
> .types(Float.class, Float.class, String.class)
> .filter(new NumberVotesFilter());
>
> //merge actors and actresses
> DataSet<Tuple3<String, String, String>> actors =
> actorsTemp.union(actresses);
> //create weighted rating
> DataSet<Tuple2<String, Float>> weightedRatings =
> ratings.map(new WeightedRatingCalculator());
>
> THIS IS WHAT I'M TRYING IN THE MAIN METHOD:
>
> actors.map(new JoinNames())
> .join(weightedRatings)
> .where(1).equalTo(0)
> .projectFirst(0).projectSecond(1)
> .groupBy(0)
> .reduceGroup(new MeanRatingCalculator())
> .first(10).print();
>
>
> And here is the GroupReduce function I wrote:
>
> public static class MeanRatingCalculator implements
> GroupReduceFunction<Tuple2<String, Float>, Tuple3<String, Float, Integer>> {
>
> public void reduce(Iterable<Tuple2<String, Float>> ratedActors,
> Collector<Tuple3<String, Float, Integer>> out) throws Exception {
>
> String name = null;
> Float ratings = 0F;
> int numberOfMovies = 0;
> for (Tuple2<String, Float> a : ratedActors) {
> //store the name
> name = a.f0;
> //update the sum of the ratings and number of movies
> ratings += a.f1;
> numberOfMovies++;
> }
> // emit name, average rating and number of films
> out.collect(new Tuple3<String, Float, Integer>(name,
> ratings/(float)numberOfMovies, numberOfMovies));
> }
> }
>
>
> I get the following error message when I try to compile the code:
>
> java: method reduceGroup in class
> org.apache.flink.api.java.operators.UnsortedGrouping<T> cannot be
> applied to given types;
> required:
> org.apache.flink.api.common.functions.GroupReduceFunction<org.apache.flink.api.java.tuple.Tuple,R>
> found: de.tub.dima.TopActors.MeanRatingCalculator
> reason: no instance(s) of type variable(s) R exist so that argument type
> de.tub.dima.TopActors.MeanRatingCalculator conforms to formal parameter
> type
> org.apache.flink.api.common.functions.GroupReduceFunction<org.apache.flink.api.java.tuple.Tuple,R>
>
>
> I can't figure out what the problem might be and would be very grateful
> for any help!! I hope I have given all the necessary information. I'm
> using Ubuntu 14.04 and IntelliJ Idea as IDE.
>
> Thank you very much,
>
> Lea