You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/02/11 11:49:29 UTC
[27/50] [abbrv] flink git commit: [FLINK-1201] [gelly] filtering bad
records and top track per user
[FLINK-1201] [gelly] filtering bad records and top track per user
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3cef2105
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3cef2105
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3cef2105
Branch: refs/heads/master
Commit: 3cef2105ae4a1c6f2c2780539ad9fc7d76747e9d
Parents: 2b98beb
Author: vasia <va...@gmail.com>
Authored: Sun Dec 14 22:44:24 2014 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Feb 11 10:46:15 2015 +0100
----------------------------------------------------------------------
.../flink/graph/example/MusicProfiles.java | 102 +++++++++++++++++++
.../flink/graph/utils/Tuple3ToEdgeMap.java | 25 +++++
2 files changed, 127 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3cef2105/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/MusicProfiles.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/MusicProfiles.java b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/MusicProfiles.java
new file mode 100644
index 0000000..cd425a3
--- /dev/null
+++ b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/MusicProfiles.java
@@ -0,0 +1,102 @@
+package flink.graphs.example;
+
+import java.util.Iterator;
+
+import org.apache.flink.api.common.ProgramDescription;
+import org.apache.flink.api.common.functions.CoGroupFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.types.NullValue;
+import org.apache.flink.util.Collector;
+
+import flink.graphs.Edge;
+import flink.graphs.EdgeDirection;
+import flink.graphs.EdgesFunction;
+import flink.graphs.Graph;
+import flink.graphs.utils.Tuple3ToEdgeMap;
+
+public class MusicProfiles implements ProgramDescription {
+
+ @SuppressWarnings("serial")
+ public static void main (String [] args) throws Exception {
+
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ /** read the user-song-play triplets
+ * The format is <userID>\t<songID>\t<playcount>
+ */
+ DataSet<Tuple3<String, String, Integer>> triplets = env.readCsvFile(args[0])
+ .lineDelimiter("\n").fieldDelimiter('\t').types(String.class, String.class, Integer.class);
+
+ /**
+ * read the mismatches dataset and extract the songIDs
+ * The format is "ERROR: <songID trackID> song_title"
+ */
+ DataSet<Tuple1<String>> mismatches = env.readTextFile(args[1]).map(
+ new MapFunction<String, Tuple1<String>>() {
+ public Tuple1<String> map(String value) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+ });
+
+ // filter out the mismatches from the triplets dataset
+ DataSet<Tuple3<String, String, Integer>> validTriplets = triplets.coGroup(mismatches)
+ .where(1).equalTo(0).with(new CoGroupFunction<Tuple3<String, String, Integer>, Tuple1<String>,
+ Tuple3<String, String, Integer>>() {
+ public void coGroup(
+ Iterable<Tuple3<String, String, Integer>> triplets,
+ Iterable<Tuple1<String>> invalidSongs,
+ Collector<Tuple3<String, String, Integer>> out) {
+ if (!invalidSongs.iterator().hasNext()) {
+ // this is a valid triplet
+ out.collect(triplets.iterator().next());
+ }
+ }
+ });
+
+ // Create a user -> song weighted bipartite graph
+ // where the edge weights correspond to play counts
+ DataSet<Edge<String, Integer>> userSongEdges = validTriplets.map(
+ new Tuple3ToEdgeMap<String, Integer>());
+
+ Graph<String, NullValue, Integer> userSongGraph = Graph.create(userSongEdges, env);
+
+ // get the top track (most listened) for each user
+ DataSet<Tuple2<String, String>> usersWithTopTrack = userSongGraph.reduceOnEdges(
+ new EdgesFunction<String, Integer, String>() {
+ public Tuple2<String, String> iterateEdges(
+ Iterable<Tuple2<String, Edge<String, Integer>>> edges) {
+ int maxPlaycount = 0;
+ String userId = "";
+ String topSong = "";
+
+ final Iterator<Tuple2<String, Edge<String, Integer>>> edgesIterator =
+ edges.iterator();
+ if (edgesIterator.hasNext()) {
+ Tuple2<String, Edge<String, Integer>> first = edgesIterator.next();
+ userId = first.f0;
+ topSong = first.f1.getTarget();
+ }
+ while (edgesIterator.hasNext()) {
+ Tuple2<String, Edge<String, Integer>> edge = edgesIterator.next();
+ if (edge.f1.getValue() > maxPlaycount) {
+ maxPlaycount = edge.f1.getValue();
+ topSong = edge.f1.getTarget();
+ }
+ }
+ return new Tuple2<String, String> (userId, topSong);
+ }
+ }, EdgeDirection.OUT);
+ }
+
+ @Override
+ public String getDescription() {
+ return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3cef2105/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/utils/Tuple3ToEdgeMap.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/utils/Tuple3ToEdgeMap.java b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/utils/Tuple3ToEdgeMap.java
new file mode 100644
index 0000000..2a6cb23
--- /dev/null
+++ b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/utils/Tuple3ToEdgeMap.java
@@ -0,0 +1,25 @@
+package flink.graphs.utils;
+
+import java.io.Serializable;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.tuple.Tuple3;
+
+import flink.graphs.Edge;
+
+/**
+ * create an Edge DataSetfrom a Tuple3 dataset
+ *
+ * @param <K>
+ * @param <EV>
+ */
+public class Tuple3ToEdgeMap<K extends Comparable<K> & Serializable,
+ EV extends Serializable> implements MapFunction<Tuple3<K, K, EV>, Edge<K, EV>> {
+
+ private static final long serialVersionUID = 1L;
+
+ public Edge<K, EV> map(Tuple3<K, K, EV> tuple) {
+ return new Edge<K, EV>(tuple.f0, tuple.f1, tuple.f2);
+ }
+
+}