[jira] [Resolved] (EAGLE-66) Eagle TypeSafe Stream Processing DSL

["Hao Chen (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/EAGLE-66?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Hao Chen resolved EAGLE-66.
---------------------------
    Resolution: Fixed

> Eagle TypeSafe Stream Processing DSL
> ------------------------------------
>
>                 Key: EAGLE-66
>                 URL: https://issues.apache.org/jira/browse/EAGLE-66
>             Project: Eagle
>          Issue Type: Improvement
>    Affects Versions: 0.3.0
>            Reporter: Hao Chen
>            Assignee: Hao Chen
>             Fix For: 0.3.0
>
>
> h1. Main Features
> 1. Typesafe API: Currently the stream processing API is not type-safe (Type info are erased by stream framework), all programming interfaces for developer are faced to Object/AnyRef, which is not very friendly and extensible for framework user (application developer):
> {code}
> public void flatMap(java.util.List<Object> input, Collector<Tuple2<String, AlertAPIEntity>> outputCollector)
> {code}
> So i propose the interface as (all type information are transparent for developer, not need additional parameters, supported by Scala implicit TypeTag)
> {code}class StreamProducer[+T <: Any] extends StreamInfo with StreamProtocol[T]{code}
> * StreamInfo:  contains Stream core information including streamId, processing element id/name, entity type info (class/TypeTag)
> * StreamProtocol extends JavaStreamProtocol: contains basic Stream DSL API and java-compatible API
> {code}
> class StreamInfo  extends Serializable{
>   val id:Int = UniqueId.incrementAndGetId()
>   var name: String = null
>   var streamId:String=null
>   var parallelismNum: Int = 1
>   var inKeyed:Boolean = false
>   var outKeyed:Boolean = false
>   var keySelector:KeySelector = null
>   var typeClass:Class[_] = classOf[AnyRef]
>   @transient  implicit var typeTag:ru.TypeTag[_] = ru.typeTag[AnyRef]
> }
> {code}
> And the StreamInfo can be shared through the runtime as implicit context for execution layer as well:
> {code}
> abstract class AbstractStreamBolt[T](val fieldsNum:Int=0, val ack:Boolean = true)(implicit streamInfo:StreamInfo) extends BaseRichBolt
> {code}
> 2. KeyValue Based Structure: currently framework user (developer) have to handle with field declaration again and again, and framework and business logic are highly coupled, according to the StreamProtocol description, user should not care about framework level detail like internal data structure for Storm using List<Object> with Fields<String> which is not friendly for developer, we should make sure user focus on business logic only like:
> {code}
> env.from(tuples)
> 	.groupByKey(_.name)
> {code}
> 3. Spout grouping instead of overriding Schema: currently especially in use case like HdfsAuditLog Monitoring, if developer wants to  groupby certain key, they are forced to override Schema (specific for storm) , which is not good and un-reusable.
> 4. Environment Decoupled: currently the stream (metadata) /dag (logic) / environment (execution) are coupled with storm internal implementation, which is not good for becoming a metadata-driven pipeline framework in future, so we should decouple it.
>  
> 5. Compatible with Field-based Structure in old framework and application.
> 6. Configuration: enhanced config wrapper upon typesafe-config for supporting get/set/default and integrated with ExecutionEnvironment
> {code}
> val env = ExecutionEnvironments.get[StormExecutionEnvironment](args)
>   val streamName = env.config.get[String]("eagle.stream.name","eventStream")
>   val streamExecutorId = env.config.get[String]("eagle.stream.executor",s"${streamName}Executor")
>   env.config.set("dataSourceConfig.deserializerClass",classOf[JsonMessageDeserializer].getCanonicalName)
> {code}
> h1. Sample Application
> {code}
> case class Entity(name:String,value:Double,var inc:Int=0)
> val tuples = Seq(
> 	Entity("a", 1),
> 	Entity("a", 2),
> 	Entity("a", 3),
> 	Entity("b", 2),
> 	Entity("c", 3),
> 	Entity("d", 3)
> )
> val env = ExecutionEnvironments.get[StormExecutionEnvironment](args)
> // DAG is fully automatically aware: Entity -> Tuple2 -> Tuple3 
> env.from(tuples)
> 	.groupByKey(_.name)
> 	.map(o => {o.inc += 2;o})
> 	.filter(_.name != "b")
> 	.filter(_.name != "c")
> 	.groupByKey(o=>(o.name,o.value))
> 	.map(o => (o.name,o))
> 	.map(o => (o._1,o._2.value,o._2.inc))
> 	.foreach(println)
>  env.execute()
> {code}
> Type is transparent for developer during both DAG compiling (programming) and runtime (metadata) phases
> {code}
> 2015-12-07 15:17:19,820 INFO [main] utils.GraphPrinter$[43]: Before expanded DAG 
> { 
> 	IterableStreamProducer[Entity]_1{1} -> GroupByKeyProducer[<function1>(Entity)]_2{1} in shuffleGroup
> 	GroupByKeyProducer[<function1>(Entity)]_2{1} -> MapProducer[Entity]_3{1} in shuffleGroup
> 	MapProducer[Entity]_3{1} -> FilterProducer[Entity]_4{1} in shuffleGroup
> 	FilterProducer[Entity]_4{1} -> FilterProducer[Entity]_5{1} in shuffleGroup
> 	FilterProducer[Entity]_5{1} -> GroupByKeyProducer[<function1>(Entity)]_6{1} in shuffleGroup
> 	GroupByKeyProducer[<function1>(Entity)]_6{1} -> MapProducer[Tuple2]_7{1} in shuffleGroup
> 	MapProducer[Tuple2]_7{1} -> MapProducer[Tuple3]_8{1} in shuffleGroup
> 	MapProducer[Tuple3]_8{1} -> ForeachProducer[void]_9{1} in shuffleGroup
> }
> 2015-12-07 15:17:19,852 INFO [main] utils.GraphPrinter$[43]: After expanded DAG 
> { 
> 	IterableStreamProducer[Entity]_1{1} -> MapProducer[Entity]_3{1} in groupByKey(<function1>)
> 	MapProducer[Entity]_3{1} -> FilterProducer[Entity]_4{1} in shuffleGroup
> 	FilterProducer[Entity]_4{1} -> FilterProducer[Entity]_5{1} in shuffleGroup
> 	FilterProducer[Entity]_5{1} -> MapProducer[Tuple2]_7{1} in groupByKey(<function1>)
> 	MapProducer[Tuple2]_7{1} -> MapProducer[Tuple3]_8{1} in shuffleGroup
> 	MapProducer[Tuple3]_8{1} -> ForeachProducer[void]_9{1} in shuffleGroup
> }
> 2015-12-07 15:17:19,898 INFO [main] storm.StormTopologyCompiler$[92]: Storm topology DAG
> {
>  	Spout[IterableStreamProducer[Entity]_1]{1} -> Bolt[MapProducer[Entity]_3]{1} in groupByKey(<function1>)
> 	Bolt[MapProducer[Entity]_3 ]{1} -> Bolt[FilterProducer[Entity]_4]{1} in shuffleGroup
> 	Bolt[FilterProducer[Entity]_4 ]{1} -> Bolt[FilterProducer[Entity]_5]{1} in shuffleGroup
> 	Bolt[FilterProducer[Entity]_5 ]{1} -> Bolt[MapProducer[Tuple2]_7]{1} in groupByKey(<function1>)
> 	Bolt[MapProducer[Tuple2]_7 ]{1} -> Bolt[MapProducer[Tuple3]_8]{1} in shuffleGroup
> 	Bolt[MapProducer[Tuple3]_8 ]{1} -> Bolt[ForeachProducer[void]_9]{1} in shuffleGroup 
> }
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)