Lazy Processing of Streams

Question

I have the following problem scenario:

• I have a text file and I have to read it and split it into lines.
• Some lines might need to be dropped (according to criteria that are not fixed).
• The lines that are not dropped must be parsed into some predefined records.
• Records that are not valid must be dropped.
• Duplicate records may exist and, in such a case, they are consecutive. If duplicate / multiple records exist, only one item should be kept.
• The remaining records should be grouped according to the value contained in one field; all records belonging to the same group appear one after another (e.g. AAAABBBBCCDEEEFF and so on).
• The records of each group should be numbered (1, 2, 3, 4, ...). For each group the numbering starts from 1.
• The records must then be saved somewhere / consumed in the same order as they were produced.

I have to implement this in Java or C++.

My first idea was to define functions / methods like:

• One method to get all the lines from the file.
• One method to filter out the unwanted lines.
• One method to parse the filtered lines into valid records.
• One method to remove duplicate records.
• One method to group records and number them.

The problem is that the data I am going to read can be too big and might not fit into main memory: so I cannot just construct all these lists and apply my functions one after the other.

On the other hand, I think I do not need to fit all the data in main memory at once because once a record has been consumed all its underlying data (basically the lines of text between the previous record and the current record, and the record itself) can be disposed of.

With the little knowledge I have of Haskell I have immediately thought about some kind of lazy evaluation, in which instead of applying functions to lists that have been completely computed, I have different streams of data that are built on top of each other and, at each moment, only the needed portion of each stream is materialized in main memory.

But I have to implement this in Java or C++. So my question is which design pattern or other technique can allow me to implement this lazy processing of streams in one of these languages.

Answer 1

You should look into iterators if deciding for Java.

Write an Iterator<String> that reads a line from the file. Write a filtering iterator that accepts the above iterator in the constructor and only generate those lines you are interested in. Write a splitting Iterator<Record> that accepts a string iterator in the constructor, and splits each line into a record. And so on.

You will most likely find that you will do the processing in the "are there more?" section to get the logic right.

Answer 2

Even though this question has an accepted answer already, I have recently found another interesting solution, which I would like to share here. I would be interested to learn about other techniques for implementing streams in Java.

The first ideas for my solution came from a course on Scala in which I was able to experiment with Scala's library streams (scala.collection.immutable.Stream). The course explained that a stream in Scala is very similar to a list (sequence of cons cells) in which the tail() function / method is computed lazily: the tail sequence is computed on demand when tail() is called for the first time, and cached for further accesses.

Of course, the tail is computed lazily as well, i.e. only the first cons cell of the tail is generated when tail() is called: subsequent elements of the tail stream are generated when tail() is called on this stream, and so on.

The second part of the solution came from this question and some of its answers: the only extension I needed was to implement cons cells with lazy tails.

Based on this solution, it was easy to define the usual functions (methods) take and drop, and the higher-order functions map, filter, dropWhile, takeWhile which also work lazily on streams. With these tools the original problem of manipulating streams of strings is much easier to solve.

My Solution

I first implemented lists as indicated in the programmers question and then implemented a similar data type Stream<T>. A cons cell consists of an object of class Cons<T> containing

1. A value of type T (the head), and
2. A closure (function object) with one apply method returning the tail stream.

Here are the key definitions:

// An interface to store functions (closures) that are
// used to produce a stream's tail on demand.
public interface IStreamFunction<A>
{
  public Stream<A> apply();
}

// Wrapper object that invokes a stream function
// and then stores the result for future invocations of tail().
public class Memoizer<Stream<A>>
{
  private final IStreamFunction<A> _f;

  private Stream<A> _v;

  public Memoizer(IStreamFunction<A> f)
  {
    _f = f;

    _v = null;
  }

  public Stream<A> value()
  {
    if (_v == null)
    {
      _v = _f.apply();
    }

    return _v;
  }
}

// Stream class with two subclasses implementing the empty stream
// and a cons cell, respectively.
public abstract class Stream<A>
{
  // Private constructor cannot be called by any subclasses except inner classes.
  private Stream()
  {
  }

  public abstract A head();

  public abstract Stream<A> tail();

  public abstract boolean isEmpty();

  // Empty stream.
  public static final class Nil<A> extends Stream<A>
  {
    public Nil()
    {
    }

    public A head()
    {
      throw new NoSuchElementException("Nil.head");
    }

    public Stream<A> tail()
    {
      throw new NoSuchElementException("Nil.tail");
    }

    public boolean isEmpty()
    {
      return true;
    }
  }

  // Cons cell.
  public static final class Cons<A> extends Stream<A>
  {
    private final A _h;
    private final Memoizer<Stream<A>> _m;

    public Cons(A h, IStreamFunction<A> f)
    {
      _h = h;
      _m = new Memoizer<Stream<A>>(f);
    }

    public A head()
    {
      return _h;
    }

    public Stream<A> tail()
    {
      return _m.value();
    }

    public boolean isEmpty()
    {
      return false;
    }
  }
}