Asynchronous iterators

Asynchronous iterators for large record sets

back to top

Phantom comes packed with CQL rows asynchronous lazy iterators to help you deal with billions of records. phantom iterators are based on Play iterators with very lightweight integration. To take advantage of this functionality, you will need to add an extra dependency, namely:

libraryDependencies ++= Seq(
  "com.websudos" %% "phantom-reactivestreams" % version 
)

The functionality is identical with respect to asynchronous, lazy behaviour and available methods. For more on this, see this Play tutorial

Usage is trivial. If you want to use slice, take or drop with iterators, the partitioner needs to be ordered. This is an extremely important easy to neglect fact. Do not rely on Cassandra to guarantee that data will be retrieved in the same order you write it in as the default Murmur3Partitioner doesn't do that.

import scala.concurrent.Await
import scala.concurrent.duration._
import com.websudos.phantom.dsl._
import com.websudos.phantom.reactivestreams._


sealed class ExampleRecord3 extends CassandraTable[ConcreteExampleRecord3, ExampleModel] {

  object id extends UUIDColumn(this) with PartitionKey[UUID]
  object order_id extends LongColumn(this) with ClusteringOrder[Long] with Descending
  object timestamp extends DateTimeColumn(this) with PrimaryKey[DateTime]
  object name extends StringColumn(this) with PrimaryKey[String]
  object props extends MapColumn[String, String](this)
  object test extends OptionalIntColumn(this)

  override def fromRow(row: Row): ExampleModel = {
    ExampleModel(
      id(row),
      name(row),
      props(row),
      timestamp(row),
      test(row)
    );
  }
}

abstract class ConcreteExampleRecord3 extends ExampleRecord3 with RootConnector {
  def getRecords(start: Int, limit: Int): Future[Set[ExampleModel]] = {
    select.fetchEnumerator.slice(start, limit).collect
  }
}