gistfile1.scala

 def lag[A: Ordering : Timestamped : EntityAware]
    (events: TypedPipe[A],
     offset: Int): TypedPipe[(A, A)] = {

    events
      .groupByEntity
      .sortBy(_.timestamp)
      .mapValueStream {
        events => {
          events.sliding(2,offset)
        }
      }
      .values
      .map { case Seq(a,b) => (a,b) }
  }