THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
| Table of Contents |
|---|
Dynamical Monitoring Framework is consist of four parts:
- Pipeline Builder
- Pipeline Runner
- Pipeline Manager
- Pipeline Designer
Dynamical Pipeline Builder
Pipeline ::= config + schema + dataflow
Configuration based DSL
| Code Block |
|---|
{
config {
// pipeline configuraiton
}
schema{
${schemaName} {
${field} : ${type}
}
}
dataflow{
// 1. DAG Node definition
${ModuleType}.${streamId} {
// module confiruation
}
// 2. DAG Connector definition
${streamId} -> ${streamId} {
// Connection configuration like grouping = "shuffle"
}
}
} |
Example:
Stream Topology Definition DSL
...
| Code Block | ||
|---|---|---|
| ||
{
config {
alertExecutorConfigs {
defaultAlertExecutor {
"parallelism" : 1
"partitioner" : "org.apache.eagle.policy.DefaultPolicyPartitioner"
"needValidation" : "true"
}
}
eagleProps {
"site" : "sandbox"
"dataSource": "HADOOP"
"dataJoinPollIntervalSec" : 30
"mailHost" : "atom.corp.ebay.com"
"mailSmtpPort":"25"
"mailDebug" : "true"
"eagleService": {
"host": "localhost"
"port": 38080
"username": "admin"
"password": "secret"
}
}
dynamicConfigSource {
"enabled" : true
"initDelayMillis" : 0
"delayMillis" : 30000
}
}
dataflow {
KafkaSource.JmxStreamOne {
parallism = 1000
topic = "metric_event_1"
zkConnection = "localhost:2181"
zkConnectionTimeoutMS = 15000
consumerGroupId = "Consumer"
fetchSize = 1048586
transactionZKServers = "localhost"
transactionZKPort = 2181
transactionZKRoot = "/consumers"
transactionStateUpdateMS = 2000
deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer"
}
KafkaSource.JmxStreamTwo {
parallism = 1000
topic = "metric_event_2"
zkConnection = "localhost:2181"
zkConnectionTimeoutMS = 15000
consumerGroupId = "Consumer"
fetchSize = 1048586
transactionZKServers = "localhost"
transactionZKPort = 2181
transactionZKRoot = "/consumers"
transactionStateUpdateMS = 2000
deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer"
}
KafkaSource.JmxStreamThree{
parallism = 1000
topic = "metric_event_3"
zkConnection = "localhost:2181"
zkConnectionTimeoutMS = 15000
consumerGroupId = "Consumer"
fetchSize = 1048586
transactionZKServers = "localhost"
transactionZKPort = 2181
transactionZKRoot = "/consumers"
transactionStateUpdateMS = 2000
deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer"
}
Console.printer {}
KafkaSink.metricStore {
topic = "metric_event_persist"
}
// KafkaSink.alertStore {
// "topic" = "alert_persist"
// "bootstrap.servers" = "localhost:6667"
// }
Alert.alerter {
upStreamNames = [JmxStreamOne,JmxStreamTwo,JmxStreamThree]
alertExecutorId = defaultAlertExecutor
}
// Aggregator.aggreator {
// upStreamNames = []
// analyzerId = ""
// cepQl = ""
// strategy = ""
// }
JmxStreamOne|JmxStreamTwo|JmxStreamThree -> alerter {
grouping = shuffle
}
JmxStreamOne|JmxStreamTwo|JmxStreamThree -> metricStore {
grouping = shuffle
}
JmxStreamOne|JmxStreamTwo|JmxStreamThree -> printer {
grouping = shuffle
}
}
} |
Dynamic
...
Pipeline Runner
- Pipeline Runner CLI
...
Dynamic Pipeline Manager
| Gliffy Diagram | ||||
|---|---|---|---|---|
|
Use Case
Hadoop JMX Metric Monitoring
| Gliffy Diagram | ||||
|---|---|---|---|---|
|