@Test public void testExec2() throws IOException { Tuple input = defaultInput(); Tuple row1 = tupleFactory.newTuple(3); Tuple row2 = tupleFactory.newTuple(3); row1.set(0, "1"); row2.set(0, "1"); row1.set(1, 5); row2.set(1, 10); row1.set(2, "http: row2.set(2, "http: DataBag bag = BagFactory.getInstance().newDefaultBag(); bag.add(row1); bag.add(row2); input.set(0, bag); Tuple output = funnel.exec(input); assertNotNull(output); assertTrue(!(Boolean) output.get(0)); assertNotNull(output.get(1)); assertEquals(output.get(1), "http: }
@Override public Tuple exec(final Tuple input) throws IOException { if (input == null || input.size() != 4) { return null; } DataBag bag = (DataBag) input.get(0); Iterator<Tuple> it = bag.iterator(); Tuple funnelUrls = (Tuple) input.get(1); Tuple funnelDag = (Tuple) input.get(2); int timeframe = (Integer) input.get(3); List<String> history = new ArrayList<String>(); List<Integer> timestamps = new ArrayList<Integer>(); Node drop = null; if (funnel == null && urlMap == null) { funnel = constructFunnel(funnelDag.getAll(), funnelUrls.getAll()); } while (it.hasNext()) { Tuple t = (Tuple) it.next(); if (t != null && t.size() == 3 && t.get(0) != null) { String url = (String) t.get(2); int timestamp = (Integer) t.get(1); history.add(url); timestamps.add(timestamp); } } Tuple output = TupleFactory.getInstance().newTuple(2); for (int i = 0; i < history.size(); i++) { String url = history.get(i); if (!url.equals(funnel.getUrl())) { continue; } Node curr = funnel; int beginning = timestamps.get(i); int end = beginning; while (i < history.size()) { i++; end = timestamps.get(i); if (end - beginning > timeframe) { break; } Node candidate = urlMap.get(history.get(i)); if (candidate == null || !curr.getChildren().contains(candidate)) { drop = curr; break; } if (candidate.getChildren().size() == 0) { output.set(0, true); output.set(1, null); return output; } if (i == history.size() - 1) { drop = candidate; break; } curr = candidate; } } System.out.println(drop); output.set(0, false); String dropUrl = (drop == null) ? null : drop.getUrl(); output.set(1, dropUrl); return output; }
Funnel extends EvalFunc<Tuple> { @Override public Tuple exec(final Tuple input) throws IOException { if (input == null || input.size() != 4) { return null; } DataBag bag = (DataBag) input.get(0); Iterator<Tuple> it = bag.iterator(); Tuple funnelUrls = (Tuple) input.get(1); Tuple funnelDag = (Tuple) input.get(2); int timeframe = (Integer) input.get(3); List<String> history = new ArrayList<String>(); List<Integer> timestamps = new ArrayList<Integer>(); Node drop = null; if (funnel == null && urlMap == null) { funnel = constructFunnel(funnelDag.getAll(), funnelUrls.getAll()); } while (it.hasNext()) { Tuple t = (Tuple) it.next(); if (t != null && t.size() == 3 && t.get(0) != null) { String url = (String) t.get(2); int timestamp = (Integer) t.get(1); history.add(url); timestamps.add(timestamp); } } Tuple output = TupleFactory.getInstance().newTuple(2); for (int i = 0; i < history.size(); i++) { String url = history.get(i); if (!url.equals(funnel.getUrl())) { continue; } Node curr = funnel; int beginning = timestamps.get(i); int end = beginning; while (i < history.size()) { i++; end = timestamps.get(i); if (end - beginning > timeframe) { break; } Node candidate = urlMap.get(history.get(i)); if (candidate == null || !curr.getChildren().contains(candidate)) { drop = curr; break; } if (candidate.getChildren().size() == 0) { output.set(0, true); output.set(1, null); return output; } if (i == history.size() - 1) { drop = candidate; break; } curr = candidate; } } System.out.println(drop); output.set(0, false); String dropUrl = (drop == null) ? null : drop.getUrl(); output.set(1, dropUrl); return output; } }
Funnel extends EvalFunc<Tuple> { @Override public Tuple exec(final Tuple input) throws IOException { if (input == null || input.size() != 4) { return null; } DataBag bag = (DataBag) input.get(0); Iterator<Tuple> it = bag.iterator(); Tuple funnelUrls = (Tuple) input.get(1); Tuple funnelDag = (Tuple) input.get(2); int timeframe = (Integer) input.get(3); List<String> history = new ArrayList<String>(); List<Integer> timestamps = new ArrayList<Integer>(); Node drop = null; if (funnel == null && urlMap == null) { funnel = constructFunnel(funnelDag.getAll(), funnelUrls.getAll()); } while (it.hasNext()) { Tuple t = (Tuple) it.next(); if (t != null && t.size() == 3 && t.get(0) != null) { String url = (String) t.get(2); int timestamp = (Integer) t.get(1); history.add(url); timestamps.add(timestamp); } } Tuple output = TupleFactory.getInstance().newTuple(2); for (int i = 0; i < history.size(); i++) { String url = history.get(i); if (!url.equals(funnel.getUrl())) { continue; } Node curr = funnel; int beginning = timestamps.get(i); int end = beginning; while (i < history.size()) { i++; end = timestamps.get(i); if (end - beginning > timeframe) { break; } Node candidate = urlMap.get(history.get(i)); if (candidate == null || !curr.getChildren().contains(candidate)) { drop = curr; break; } if (candidate.getChildren().size() == 0) { output.set(0, true); output.set(1, null); return output; } if (i == history.size() - 1) { drop = candidate; break; } curr = candidate; } } System.out.println(drop); output.set(0, false); String dropUrl = (drop == null) ? null : drop.getUrl(); output.set(1, dropUrl); return output; } Funnel(); }
Funnel extends EvalFunc<Tuple> { @Override public Tuple exec(final Tuple input) throws IOException { if (input == null || input.size() != 4) { return null; } DataBag bag = (DataBag) input.get(0); Iterator<Tuple> it = bag.iterator(); Tuple funnelUrls = (Tuple) input.get(1); Tuple funnelDag = (Tuple) input.get(2); int timeframe = (Integer) input.get(3); List<String> history = new ArrayList<String>(); List<Integer> timestamps = new ArrayList<Integer>(); Node drop = null; if (funnel == null && urlMap == null) { funnel = constructFunnel(funnelDag.getAll(), funnelUrls.getAll()); } while (it.hasNext()) { Tuple t = (Tuple) it.next(); if (t != null && t.size() == 3 && t.get(0) != null) { String url = (String) t.get(2); int timestamp = (Integer) t.get(1); history.add(url); timestamps.add(timestamp); } } Tuple output = TupleFactory.getInstance().newTuple(2); for (int i = 0; i < history.size(); i++) { String url = history.get(i); if (!url.equals(funnel.getUrl())) { continue; } Node curr = funnel; int beginning = timestamps.get(i); int end = beginning; while (i < history.size()) { i++; end = timestamps.get(i); if (end - beginning > timeframe) { break; } Node candidate = urlMap.get(history.get(i)); if (candidate == null || !curr.getChildren().contains(candidate)) { drop = curr; break; } if (candidate.getChildren().size() == 0) { output.set(0, true); output.set(1, null); return output; } if (i == history.size() - 1) { drop = candidate; break; } curr = candidate; } } System.out.println(drop); output.set(0, false); String dropUrl = (drop == null) ? null : drop.getUrl(); output.set(1, dropUrl); return output; } Funnel(); @Override /** * {@inheritDoc} * * Method exec takes a tuple containing four objects: * 1) A bag of tuples containing: the user id, the time stamp, the uri * request * 2) A tuple of URLs * 3) An integer as the maximum timeframe between the first and last * request in the funnel in seconds. * * Doing the sorting in Pig Latin, rather than in your UDF, is important * for a couple of reasons. One, it means Pig can offload the sorting to * MapReduce. MapReduce has the ability to sort data by a secondary key * while grouping it. So the order statement in this case does not * require a separate sorting operation. Two, it means that your UDF * does not need to wait for all data to be available before it starts * processing. Instead, it can use the Accumulator interface (see the * section called “Accumulator Interface”) which is much more memory * efficient. */ Tuple exec(final Tuple input); }
Funnel extends EvalFunc<Tuple> { @Override public Tuple exec(final Tuple input) throws IOException { if (input == null || input.size() != 4) { return null; } DataBag bag = (DataBag) input.get(0); Iterator<Tuple> it = bag.iterator(); Tuple funnelUrls = (Tuple) input.get(1); Tuple funnelDag = (Tuple) input.get(2); int timeframe = (Integer) input.get(3); List<String> history = new ArrayList<String>(); List<Integer> timestamps = new ArrayList<Integer>(); Node drop = null; if (funnel == null && urlMap == null) { funnel = constructFunnel(funnelDag.getAll(), funnelUrls.getAll()); } while (it.hasNext()) { Tuple t = (Tuple) it.next(); if (t != null && t.size() == 3 && t.get(0) != null) { String url = (String) t.get(2); int timestamp = (Integer) t.get(1); history.add(url); timestamps.add(timestamp); } } Tuple output = TupleFactory.getInstance().newTuple(2); for (int i = 0; i < history.size(); i++) { String url = history.get(i); if (!url.equals(funnel.getUrl())) { continue; } Node curr = funnel; int beginning = timestamps.get(i); int end = beginning; while (i < history.size()) { i++; end = timestamps.get(i); if (end - beginning > timeframe) { break; } Node candidate = urlMap.get(history.get(i)); if (candidate == null || !curr.getChildren().contains(candidate)) { drop = curr; break; } if (candidate.getChildren().size() == 0) { output.set(0, true); output.set(1, null); return output; } if (i == history.size() - 1) { drop = candidate; break; } curr = candidate; } } System.out.println(drop); output.set(0, false); String dropUrl = (drop == null) ? null : drop.getUrl(); output.set(1, dropUrl); return output; } Funnel(); @Override /** * {@inheritDoc} * * Method exec takes a tuple containing four objects: * 1) A bag of tuples containing: the user id, the time stamp, the uri * request * 2) A tuple of URLs * 3) An integer as the maximum timeframe between the first and last * request in the funnel in seconds. * * Doing the sorting in Pig Latin, rather than in your UDF, is important * for a couple of reasons. One, it means Pig can offload the sorting to * MapReduce. MapReduce has the ability to sort data by a secondary key * while grouping it. So the order statement in this case does not * require a separate sorting operation. Two, it means that your UDF * does not need to wait for all data to be available before it starts * processing. Instead, it can use the Accumulator interface (see the * section called “Accumulator Interface”) which is much more memory * efficient. */ Tuple exec(final Tuple input);  }