KeyedPipe

[Supun Kamburugamuve]

Hi Ahmet,

I found that the API is correct for KeyedPipe. The function connecting to the pipe can only use streaming semantics. So the code looks like following

// calculate matched tweetID-datepairs

CachedTSet<Tuple<String, BigInteger>> cachedMatchedTweets =

    persistedTweets
        .keyedPipe()
        .compute(new ComputeMatchingTweets())
        .addInput("delete-input", cachedDeleteIDs)
        .pipe()
        .cache();

private class ComputeMatchingTweets extends BaseComputeCollectorFunc<Tuple<String, BigInteger>, Tuple<BigInteger, String>> {

  private TSetContext ctx;
  private Set<BigInteger> deleteSet = new TreeSet<>();

  public ComputeMatchingTweets() {
  }

  @Override
  public void prepare(TSetContext context) {
    ctx = context;

    DataPartition a = context.getInput("delete-input");
    DataPartitionConsumer<BigInteger> consumer = a.getConsumer();
    while (consumer.hasNext()) {
      BigInteger deleteTweetID = consumer.next();
      deleteSet.add(deleteTweetID);
    }
    LOG.info("ADDED TUPLES TO DELETE SET: " + deleteSet.size());
  }

  @Override
  public void compute(Tuple<BigInteger, String> input, RecordCollector<Tuple<String, BigInteger>> output) {
    if (deleteSet.contains(input.getKey())) {
      output.collect(new Tuple<>(input.getValue(), input.getKey()));
    }
  }
}

--

Supun Kamburugamuve, PhD

Digital Science Center, Indiana University

Member, Apache Software Foundation; http://www.apache.org
E-mail: supun@apache.org;  Mobile: +1 812 219 2563

[Ahmet Uyar]

Hi Supun,

Thanks for that. It is working now with pipe. 

keyedPipe and keyedDirect works with different data types then. 

keyedDirect expects an iterator but keyedPipe expects a tuple. 

I thought both would work with the same data type.

Ahmet

[Supun Kamburugamuve]

Hi Ahmet,

Pipe is an in-memory streaming semantic. So it cannot have iterator afterwords.

task1 -> pipe -> task2

For pipe, we assume task1 and task2 are in-memory (most of the time). Task 1 can be producing multiple tuples and these are streamed to the task2. 

Best,

Supun..