[
https://issues.apache.org/jira/browse/EAGLE-66?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15055717#comment-15055717
]
ASF GitHub Bot commented on EAGLE-66:
-------------------------------------
Github user RalphSu commented on a diff in the pull request:
https://github.com/apache/incubator-eagle/pull/26#discussion_r47477417
--- Diff:
eagle-core/eagle-data-process/eagle-stream-process-api/src/test/java/org/apache/eagle/datastream/TestJavaReflectionUtils.java
---
@@ -0,0 +1,19 @@
+package org.apache.eagle.datastream;
--- End diff --
+1 for license missing
> 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)
