LambdaFlow: Scala Functional Message Processing

Scala Functional Message Processing: LambdaFlow John Nestor 47 Degrees www.47deg.com https://github.com/47deg/LambdaFlow May 9, 2017 147deg.com

47deg.com © Copyright 2017 47 Degrees Inspiration • Wanted to build a reactive system using Akka clustering • Don’t like Spark micro-batching, want message at a time • Want a Scala functional DSL, like the Spark Scala DSL 4

47deg.com © Copyright 2017 47 Degrees LambdaFlow • Near-real-time message processing • Message at a time • Both set (create or update) and delete • Written in Scala and Akka • Two parts: • Functional DSL • Akka Cluster reactive implementation 5

47deg.com © Copyright 2017 47 Degrees Combining Actions • Most actions have • Input: used to set or delete keys in KV store • Output: used to set or delete keys on next action • Operation: maps KV of this action to KV of next action 10

47deg.com © Copyright 2017 47 Degrees Functional Eventual Consistency • Two Actions A1[K1,V1] and A2[K2,V2] • Each have a KV store: A1.kv[K1,V1] and A2.kv[K2,V2] • the output of A1 is the input to A2 • A1 has an operation A1.f that maps   [K1,V1] => Set([K2,V2]) • If all changes to A1 are stopped, eventually • A2.kv = A1.kv.ﬂatMap(A1.f) 11

47deg.com © Copyright 2017 47 Degrees Incremental Operation Example • Suppose action KV type is String=>String • If we change “A”->”B” to “A”->”C” • Operation example 1 (map): (k,v)=>(k+v,v+v) • Delete “AB”->”BB” • Set “AC”=> “CC” • Operation example 2(mapvalue): (k,v) => (k,v+v) • Delete “A” -> “BB” (not needed here) • Set “A” -> “CC” 13

47deg.com © Copyright 2017 47 Degrees Implementations • Class based • Uses method calls rather than actor messages • Local Akka Actor based • Akka Cluster Based • Multinode • Reactive: scalable, responsive, fault-tolerant 14

47deg.com © Copyright 2017 47 Degrees Simple Example Code val src = Source[String, String]("src")  val counts = Sink[String, Long]("counts")    val f1 = src  .map("map",(uid,lang)=>((uid,lang),()))  .reduce("red", _._2, (k, v) => 1L, LongSum)  .sink(counts)    run(f1) 18

47deg.com © Copyright 2017 47 Degrees Source and Sink Actions • Source[K,V] • gets input from outside the graph (e.g. Kafka) • outputs to other actions • Sink[K,V] • gets it input from other actions • outputs to outside the graph (e.g. Kafka) 19

47deg.com © Copyright 2017 47 Degrees Reduce Operation • Action input type K,V • Action output type KOUT,VOUT • Grouping function type: K=>KOUT (many to one) • Value function type: (K,V)=>VOUT • Reduce operation type (VOUT,VOUT)=>VOUT • commutative, associative and inverse (abelian group) 21

47deg.com © Copyright 2017 47 Degrees Windowing Example • Streaming Data • 1000 Tweets • sent in at ~20 per second • each tweet expires after 20 seconds • Count languages and ﬁnd 20 most common words • Uses local actor implementation • Uses Vaadin based visualization tool 24

47deg.com © Copyright 2017 47 Degrees Windowing Example Code val f1 = src  .reduce("red", _._1, (k, v) => 1L, LongSum)  .to(counts1, counts2)  .sink(counts)    val f2 = counts1  .reduce("red1", _ => List.empty[String], (k, v) => v, LongSum)  .map[String, Long]("map1", (k, v) => ("total", v))  .to(total1)  .sink(total)    val f3 = counts2  .join[String, Long, Unit, String, String]("product", total1, (s:String)=>(), (s:String)=>()) {  case (lang, count, t, total) => Set((lang, f"${count * 100 / total}%02d$PERCENT"))  }  .sink(percent)    val f4 = text  .flatMap("GetWords", (k, v) => {  val (text, uuid) = k  getWords(text)  .zipWithIndex  .map {  case (word, i) => ((word, uuid, i), ())  }  .toSet  })  .map("CountOrder", (word, count) => ((-count, word), ()))  .first("first", 20)  .map("CountOrder1", (k, v) => (k._2, f"${-k._1}%4d"))  .sink(words) run(f1, f2, f3, f4) 26

47deg.com © Copyright 2017 47 Degrees Labels,To and From • Label[K,V]: used to connect an action output to another action input • .to[K,V]: send one input to multiple outputs (broadcast) • .from[K,V]: combine multiple inputs to a single output (union) 27

47deg.com © Copyright 2017 47 Degrees First Action • Action input type K,V • Action output type K,V • Count: a small integer • Output at most count key-values • those with the lowest keys • requires type K be ordered 29

47deg.com © Copyright 2017 47 Degrees Join Action • Two inputs: K1,V1 and K2,V2 • Two key value stores • Output types KOUT and VOUT • Two mapping functions: K1=>K K2=>K (many to one) • An operation • Applied to each member of the product of KV pairs from the two key stores with the same K value • (K1,V1,K2,V2) => Set[(KOUT,VOUT)] 30

47deg.com © Copyright 2017 47 Degrees Index and Query Example • Input data: Breweries • look at breweries.data • Build indexes • set of breweries at a location (State/Country) • set of breweries with a key word (in its description) • use jline3 command hander 34

47deg.com © Copyright 2017 47 Degrees Query • words:Sink[String,Set[Int]] • words.get(word) => Future[Option[Set[Id]]] • breweries:Source[Int,String] : id->JsonString • def name(json:String)=>name (extracts the brewery name) • breweries.get(id,name) => Future[Option[String]] 36

47deg.com © Copyright 2017 47 Degrees Graph Example • Note: there are two graphs • The input data • The graph of actions • Output • Number of distinct paths between two nodes • Set of all cycles • Loop in action graph for transitive closure • Like recursive function • Functional semantics is ﬁxed point of loop 39

47deg.com © Copyright 2017 47 Degrees Graph Transitive Closure • Given • route C -> E -> F • link A -> C • Form product (Common key = C) • route C -> E -> F (routes starting at C) • ﬂipped link C -> A (links ending at C) • Result • route A -> C -> E -> F (goes back around loop!) 40

47deg.com © Copyright 2017 47 Degrees LambdaFlow and Kafka Common Properties • JVM based • Integrate with Kafka topics • Key-value based • One message at a time • Operations: map, reduce, … • Scalable and fault tolerant • Open source 56

LambdaFlow: Scala Functional Message Processing

More Related Content

What's hot

Similar to LambdaFlow: Scala Functional Message Processing

More from John Nestor

Recently uploaded

LambdaFlow: Scala Functional Message Processing