eagle-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jinh...@apache.org
Subject incubator-eagle git commit: [EAGLE-840] add task attempt stream for bad node detection
Date Thu, 15 Dec 2016 12:03:09 GMT
Repository: incubator-eagle
Updated Branches:
  refs/heads/master bb71d37df -> 8fe968cb7


[EAGLE-840] add task attempt stream for bad node detection

Author: wujinhu <wujinhu920@126.com>

Closes #744 from wujinhu/EAGLE-840.


Project: http://git-wip-us.apache.org/repos/asf/incubator-eagle/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-eagle/commit/8fe968cb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-eagle/tree/8fe968cb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-eagle/diff/8fe968cb

Branch: refs/heads/master
Commit: 8fe968cb7e340e2f21711b5a0f701e760cb91d59
Parents: bb71d37
Author: wujinhu <wujinhu920@126.com>
Authored: Thu Dec 15 20:03:02 2016 +0800
Committer: wujinhu <wujinhu920@126.com>
Committed: Thu Dec 15 20:03:02 2016 +0800

----------------------------------------------------------------------
 .../siddhiext/StringEmptyFunctionExtension.java | 86 ++++++++++++++++++++
 .../src/main/resources/string.siddhiext         | 18 ++++
 .../jpm/mr/history/MRHistoryJobApplication.java | 33 ++++++--
 .../history/crawler/EagleOutputCollector.java   |  2 +
 .../JobHistorySpoutCollectorInterceptor.java    |  5 ++
 .../mr/history/parser/JHFEventReaderBase.java   | 19 ++---
 .../JobEntityCreationEagleServiceListener.java  | 22 ++---
 .../history/publisher/JobStreamPublisher.java   | 54 ++++++++++++
 .../mr/history/publisher/StreamPublisher.java   | 39 +++++++++
 .../publisher/StreamPublisherManager.java       | 54 ++++++++++++
 .../publisher/TaskAttemptStreamPublisher.java   | 54 ++++++++++++
 .../jpm/mr/history/storm/JobHistorySpout.java   | 15 ++--
 ....history.MRHistoryJobApplicationProvider.xml | 81 ++++++++++++++++--
 .../src/main/resources/application.conf         |  3 +-
 14 files changed, 443 insertions(+), 42 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8fe968cb/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringEmptyFunctionExtension.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringEmptyFunctionExtension.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringEmptyFunctionExtension.java
new file mode 100644
index 0000000..e047662
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringEmptyFunctionExtension.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eagle.alert.siddhiext;
+
+import org.wso2.siddhi.core.config.ExecutionPlanContext;
+import org.wso2.siddhi.core.executor.ExpressionExecutor;
+import org.wso2.siddhi.core.executor.function.FunctionExecutor;
+import org.wso2.siddhi.query.api.definition.Attribute;
+import org.wso2.siddhi.query.api.exception.ExecutionPlanValidationException;
+
+public class StringEmptyFunctionExtension extends FunctionExecutor {
+    /**
+     * The initialization method for StringEmptyFunctionExtension, this method will be called
before the other methods.
+     *
+     * @param attributeExpressionExecutors the executors of each function parameter
+     * @param executionPlanContext         the context of the execution plan
+     */
+    @Override
+    protected void init(ExpressionExecutor[] attributeExpressionExecutors, ExecutionPlanContext
executionPlanContext) {
+        if (attributeExpressionExecutors.length != 1) {
+            throw new ExecutionPlanValidationException("Invalid no of arguments passed to
string:empty() function, "
+                    + "required 1, but found " + attributeExpressionExecutors.length);
+        }
+
+        Attribute.Type attributeType = attributeExpressionExecutors[0].getReturnType();
+        if (attributeType != Attribute.Type.STRING) {
+            throw new ExecutionPlanValidationException("Invalid parameter type found for
the argument of math:string() function, "
+                    + "required " + Attribute.Type.STRING
+                    + ", but found " + attributeType.toString());
+        }
+    }
+
+    /**
+     * The main execution method which will be called upon event arrival.
+     * when there are more than one function parameter
+     *
+     * @param data the runtime values of function parameters
+     * @return the function result
+     */
+    @Override
+    protected Object execute(Object[] data) {
+        return null;
+    }
+
+    @Override
+    protected Object execute(Object data) {
+        return !(data == null || ((String) data).isEmpty());
+    }
+
+    @Override
+    public void start() {
+    }
+
+    @Override
+    public void stop() {
+    }
+
+    @Override
+    public Attribute.Type getReturnType() {
+        return Attribute.Type.BOOL;
+    }
+
+    @Override
+    public Object[] currentState() {
+        return null;
+    }
+
+    @Override
+    public void restoreState(Object[] state) {
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8fe968cb/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/resources/string.siddhiext
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/resources/string.siddhiext
b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/resources/string.siddhiext
new file mode 100644
index 0000000..e16be99
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/resources/string.siddhiext
@@ -0,0 +1,18 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+empty=org.apache.eagle.alert.siddhiext.StringEmptyFunctionExtension
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8fe968cb/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/MRHistoryJobApplication.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/MRHistoryJobApplication.java
b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/MRHistoryJobApplication.java
index 66906f0..e2695a3 100644
--- a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/MRHistoryJobApplication.java
+++ b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/MRHistoryJobApplication.java
@@ -22,6 +22,10 @@ import org.apache.eagle.app.environment.impl.StormEnvironment;
 import org.apache.eagle.app.messaging.StormStreamSink;
 import org.apache.eagle.jpm.mr.history.crawler.JobHistoryContentFilter;
 import org.apache.eagle.jpm.mr.history.crawler.JobHistoryContentFilterBuilder;
+import org.apache.eagle.jpm.mr.history.crawler.JobHistorySpoutCollectorInterceptor;
+import org.apache.eagle.jpm.mr.history.publisher.JobStreamPublisher;
+import org.apache.eagle.jpm.mr.history.publisher.StreamPublisherManager;
+import org.apache.eagle.jpm.mr.history.publisher.TaskAttemptStreamPublisher;
 import org.apache.eagle.jpm.mr.history.storm.JobHistorySpout;
 import org.apache.eagle.jpm.util.Constants;
 
@@ -63,17 +67,36 @@ public class MRHistoryJobApplication extends StormApplication {
         TopologyBuilder topologyBuilder = new TopologyBuilder();
         String spoutName = "mrHistoryJobSpout";
         int tasks = jhfAppConf.getInt("stormConfig.mrHistoryJobSpoutTasks");
+        JobHistorySpoutCollectorInterceptor collectorInterceptor = new JobHistorySpoutCollectorInterceptor();
+        JobHistorySpout jobHistorySpout = new JobHistorySpout(filter, appConfig, collectorInterceptor);
         topologyBuilder.setSpout(
                 spoutName,
-                new JobHistorySpout(filter, appConfig),
+                jobHistorySpout,
                 tasks
         ).setNumTasks(tasks);
 
-        StormStreamSink sinkBolt = environment.getStreamSink("mr_failed_job_stream", config);
-        BoltDeclarer kafkaBoltDeclarer = topologyBuilder.setBolt("HistoryKafkaSink", sinkBolt,
jhfAppConf.getInt("stormConfig.historyKafkaSinkTasks"))
-                .setNumTasks(jhfAppConf.getInt("stormConfig.historyKafkaSinkTasks"));
-        kafkaBoltDeclarer.shuffleGrouping(spoutName);
+        StormStreamSink jobSinkBolt = environment.getStreamSink("MAP_REDUCE_JOB_STREAM",
config);
+        String jobSinkBoltName = "JobKafkaSink";
+        BoltDeclarer jobKafkaBoltDeclarer = topologyBuilder.setBolt(jobSinkBoltName, jobSinkBolt,
jhfAppConf.getInt("stormConfig.jobKafkaSinkTasks"))
+                .setNumTasks(jhfAppConf.getInt("stormConfig.jobKafkaSinkTasks"));
+        String spoutToJobSinkName = spoutName + "_to_" + jobSinkBoltName;
+        jobKafkaBoltDeclarer.shuffleGrouping(spoutName, spoutToJobSinkName);
 
+        StormStreamSink taskAttemptSinkBolt = environment.getStreamSink("MAP_REDUCE_TASK_ATTEMPT_STREAM",
config);
+        String taskAttemptSinkBoltName = "TaskAttemptKafkaSink";
+        BoltDeclarer taskAttemptKafkaBoltDeclarer = topologyBuilder.setBolt(taskAttemptSinkBoltName,
taskAttemptSinkBolt, jhfAppConf.getInt("stormConfig.taskAttemptKafkaSinkTasks"))
+                .setNumTasks(jhfAppConf.getInt("stormConfig.taskAttemptKafkaSinkTasks"));
+        String spoutToTaskAttemptSinkName = spoutName + "_to_" + taskAttemptSinkBoltName;
+        taskAttemptKafkaBoltDeclarer.shuffleGrouping(spoutName, spoutToTaskAttemptSinkName);
+
+        List<String> streams = new ArrayList<>();
+        streams.add(spoutToJobSinkName);
+        streams.add(spoutToTaskAttemptSinkName);
+        jobHistorySpout.setStreams(streams);
+
+        //4, add stream publisher
+        StreamPublisherManager.getInstance().addStreamPublisher(new JobStreamPublisher(spoutToJobSinkName,
collectorInterceptor));
+        StreamPublisherManager.getInstance().addStreamPublisher(new TaskAttemptStreamPublisher(spoutToTaskAttemptSinkName,
collectorInterceptor));
         return topologyBuilder.createTopology();
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8fe968cb/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/crawler/EagleOutputCollector.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/crawler/EagleOutputCollector.java
b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/crawler/EagleOutputCollector.java
index 70eab38..e4f1de8 100644
--- a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/crawler/EagleOutputCollector.java
+++ b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/crawler/EagleOutputCollector.java
@@ -26,4 +26,6 @@ import java.io.Serializable;
  */
 public interface EagleOutputCollector extends Serializable {
     void collect(ValuesArray t);
+
+    void collect(String streamId, ValuesArray t);
 }

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8fe968cb/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/crawler/JobHistorySpoutCollectorInterceptor.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/crawler/JobHistorySpoutCollectorInterceptor.java
b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/crawler/JobHistorySpoutCollectorInterceptor.java
index a7dc9a8..fac281f 100644
--- a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/crawler/JobHistorySpoutCollectorInterceptor.java
+++ b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/crawler/JobHistorySpoutCollectorInterceptor.java
@@ -32,4 +32,9 @@ public class JobHistorySpoutCollectorInterceptor implements EagleOutputCollector
     public void collect(ValuesArray t) {
         collector.emit(t);
     }
+
+    @Override
+    public void collect(String steamId, ValuesArray t) {
+        collector.emit(steamId, t);
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8fe968cb/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/parser/JHFEventReaderBase.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/parser/JHFEventReaderBase.java
b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/parser/JHFEventReaderBase.java
index fb05bae..d89937e 100644
--- a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/parser/JHFEventReaderBase.java
+++ b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/parser/JHFEventReaderBase.java
@@ -463,18 +463,17 @@ public abstract class JHFEventReaderBase extends JobEntityCreationPublisher
impl
             entityCreated(entity);
             attempt2ErrorMsg.put(taskAttemptID, Pair.of(taskID, entity.getError()));
             //generate TaskAttemptErrorCategoryEntity
-            TaskAttemptErrorCategoryEntity taskAttemptErrorCategoryEntity = new TaskAttemptErrorCategoryEntity();
-            Map<String, String> taskAttemptErrorCategoryEntityTags = new HashMap<>(entity.getTags());
-            taskAttemptErrorCategoryEntity.setTags(taskAttemptErrorCategoryEntityTags);
-            if (!taskAttemptErrorCategoryEntityTags.containsKey(MRJobTagName.ERROR_CATEGORY.toString()))
{
-                taskAttemptErrorCategoryEntityTags.put(MRJobTagName.ERROR_CATEGORY.toString(),
"");
+            if (entity.getTags().containsKey(MRJobTagName.ERROR_CATEGORY.toString())) {
+                TaskAttemptErrorCategoryEntity taskAttemptErrorCategoryEntity = new TaskAttemptErrorCategoryEntity();
+                Map<String, String> taskAttemptErrorCategoryEntityTags = new HashMap<>(entity.getTags());
+                taskAttemptErrorCategoryEntity.setTags(taskAttemptErrorCategoryEntityTags);
+
+                taskAttemptErrorCategoryEntity.setStartTime(entity.getStartTime());
+                taskAttemptErrorCategoryEntity.setEndTime(entity.getEndTime());
+                taskAttemptErrorCategoryEntity.setTimestamp(entity.getTimestamp());
+                entityCreated(taskAttemptErrorCategoryEntity);
             }
 
-            taskAttemptErrorCategoryEntity.setStartTime(entity.getStartTime());
-            taskAttemptErrorCategoryEntity.setEndTime(entity.getEndTime());
-            taskAttemptErrorCategoryEntity.setTimestamp(entity.getTimestamp());
-            entityCreated(taskAttemptErrorCategoryEntity);
-
             taskAttemptStartTime.remove(taskAttemptID);
         } else {
             // silently ignore

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8fe968cb/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/parser/JobEntityCreationEagleServiceListener.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/parser/JobEntityCreationEagleServiceListener.java
b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/parser/JobEntityCreationEagleServiceListener.java
index 4eb58a7..0c7f8c8 100644
--- a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/parser/JobEntityCreationEagleServiceListener.java
+++ b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/parser/JobEntityCreationEagleServiceListener.java
@@ -22,6 +22,8 @@ import org.apache.eagle.dataproc.impl.storm.ValuesArray;
 import org.apache.eagle.jpm.mr.history.MRHistoryJobConfig;
 import org.apache.eagle.jpm.mr.history.crawler.EagleOutputCollector;
 import org.apache.eagle.jpm.mr.history.metrics.JobExecutionMetricsCreationListener;
+import org.apache.eagle.jpm.mr.history.publisher.StreamPublisher;
+import org.apache.eagle.jpm.mr.history.publisher.StreamPublisherManager;
 import org.apache.eagle.jpm.mr.history.zkres.JobHistoryZKStateManager;
 import org.apache.eagle.jpm.mr.historyentity.*;
 import org.apache.eagle.jpm.mr.historyentity.JobExecutionAPIEntity;
@@ -109,8 +111,9 @@ public class JobEntityCreationEagleServiceListener implements HistoryJobEntityCr
                     ((JobExecutionAPIEntity) entity).getCurrentState());
 
                 metricEntities.addAll(jobExecutionMetricsCreationListener.generateMetrics((JobExecutionAPIEntity)entity));
-                if (((JobExecutionAPIEntity)entity).getCurrentState().equals(Constants.JobState.FAILED.toString()))
{
-                    emitFailedJob((JobExecutionAPIEntity) entity);
+                StreamPublisher streamPublisher = StreamPublisherManager.getInstance().getStreamPublisher(JobExecutionAPIEntity.class);
+                if (streamPublisher != null) {
+                    streamPublisher.flush((JobExecutionAPIEntity) entity);
                 }
             } else if (entity instanceof JobEventAPIEntity) {
                 jobEvents.add((JobEventAPIEntity) entity);
@@ -118,6 +121,10 @@ public class JobEntityCreationEagleServiceListener implements HistoryJobEntityCr
                 taskExecs.add((TaskExecutionAPIEntity) entity);
             } else if (entity instanceof TaskAttemptExecutionAPIEntity) {
                 taskAttemptExecs.add((TaskAttemptExecutionAPIEntity) entity);
+                StreamPublisher streamPublisher = StreamPublisherManager.getInstance().getStreamPublisher(TaskAttemptExecutionAPIEntity.class);
+                if (streamPublisher != null) {
+                    streamPublisher.flush((TaskAttemptExecutionAPIEntity) entity);
+                }
             } else if (entity instanceof TaskAttemptErrorCategoryEntity) {
                 taskAttemptErrors.add((TaskAttemptErrorCategoryEntity) entity);
             }
@@ -172,15 +179,4 @@ public class JobEntityCreationEagleServiceListener implements HistoryJobEntityCr
             throw new Exception("Entity creation fails going to EagleService");
         }
     }
-
-    private void emitFailedJob(JobExecutionAPIEntity entity) {
-        Map<String, Object> fields = new HashMap<>(entity.getTags());
-        fields.put("submissionTime", entity.getSubmissionTime());
-        fields.put("startTime", entity.getStartTime());
-        fields.put("endTime", entity.getEndTime());
-        fields.put("currentState", entity.getCurrentState());
-        fields.put("trackingUrl", entity.getTrackingUrl());
-
-        collector.collect(new ValuesArray(fields.get(MRJobTagName.JOB_ID.toString()), fields));
-    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8fe968cb/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/JobStreamPublisher.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/JobStreamPublisher.java
b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/JobStreamPublisher.java
new file mode 100644
index 0000000..06f1ff2
--- /dev/null
+++ b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/JobStreamPublisher.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+package org.apache.eagle.jpm.mr.history.publisher;
+
+import org.apache.eagle.dataproc.impl.storm.ValuesArray;
+import org.apache.eagle.jpm.mr.history.crawler.EagleOutputCollector;
+import org.apache.eagle.jpm.mr.historyentity.JobExecutionAPIEntity;
+import org.apache.eagle.jpm.util.MRJobTagName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class JobStreamPublisher extends StreamPublisher<JobExecutionAPIEntity> {
+    private static final Logger LOG = LoggerFactory.getLogger(JobStreamPublisher.class);
+
+    public JobStreamPublisher(String stormStreamId, EagleOutputCollector collector) {
+        super(stormStreamId, collector);
+    }
+
+    @Override
+    public Class<?> type() {
+        return JobExecutionAPIEntity.class;
+    }
+
+    @Override
+    public void flush(JobExecutionAPIEntity entity) {
+        Map<String, Object> fields = new HashMap<>(entity.getTags());
+        fields.put("submissionTime", entity.getSubmissionTime());
+        fields.put("startTime", entity.getStartTime());
+        fields.put("endTime", entity.getEndTime());
+        fields.put("currentState", entity.getCurrentState());
+        fields.put("trackingUrl", entity.getTrackingUrl());
+
+        collector.collect(stormStreamId, new ValuesArray(fields.get(MRJobTagName.JOB_ID.toString()),
fields));
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8fe968cb/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/StreamPublisher.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/StreamPublisher.java
b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/StreamPublisher.java
new file mode 100644
index 0000000..17eb922
--- /dev/null
+++ b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/StreamPublisher.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+package org.apache.eagle.jpm.mr.history.publisher;
+
+import org.apache.eagle.jpm.mr.history.crawler.EagleOutputCollector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class StreamPublisher<T> {
+    private static final Logger LOG = LoggerFactory.getLogger(StreamPublisher.class);
+
+    protected String stormStreamId;
+    protected EagleOutputCollector collector;
+
+    public StreamPublisher(String stormStreamId, EagleOutputCollector collector) {
+        this.stormStreamId = stormStreamId;
+        this.collector = collector;
+    }
+
+    public abstract Class<?> type();
+
+    public abstract void flush(T entity);
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8fe968cb/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/StreamPublisherManager.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/StreamPublisherManager.java
b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/StreamPublisherManager.java
new file mode 100644
index 0000000..b9b68c3
--- /dev/null
+++ b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/StreamPublisherManager.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+package org.apache.eagle.jpm.mr.history.publisher;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+
+public class StreamPublisherManager implements Serializable {
+    private static final Logger LOG = LoggerFactory.getLogger(StreamPublisherManager.class);
+
+    private static StreamPublisherManager instance = new StreamPublisherManager();
+
+    //Entity Class, StreamPublisher, currently, each Entity Class has only one publisher
+    private Map<Class<?>, StreamPublisher> publishers = new HashMap<>();
+
+    public static StreamPublisherManager getInstance() {
+        return instance;
+    }
+
+    public void addStreamPublisher(StreamPublisher streamPublisher) {
+        if (publishers.containsKey(streamPublisher.type())) {
+            LOG.warn("stream {} publisher already exists", streamPublisher.type().toString());
+        }
+
+        publishers.put(streamPublisher.type(), streamPublisher);
+    }
+
+    public StreamPublisher getStreamPublisher(Class<?> type) {
+        if (!publishers.containsKey(type)) {
+            return null;
+        }
+        return publishers.get(type);
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8fe968cb/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/TaskAttemptStreamPublisher.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/TaskAttemptStreamPublisher.java
b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/TaskAttemptStreamPublisher.java
new file mode 100644
index 0000000..e196604
--- /dev/null
+++ b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/publisher/TaskAttemptStreamPublisher.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+package org.apache.eagle.jpm.mr.history.publisher;
+
+import org.apache.eagle.dataproc.impl.storm.ValuesArray;
+import org.apache.eagle.jpm.mr.history.crawler.EagleOutputCollector;
+import org.apache.eagle.jpm.mr.historyentity.TaskAttemptExecutionAPIEntity;
+import org.apache.eagle.jpm.util.MRJobTagName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class TaskAttemptStreamPublisher extends StreamPublisher<TaskAttemptExecutionAPIEntity>
{
+    private static final Logger LOG = LoggerFactory.getLogger(TaskAttemptStreamPublisher.class);
+
+    public TaskAttemptStreamPublisher(String stormStreamId, EagleOutputCollector collector)
{
+        super(stormStreamId, collector);
+    }
+
+    @Override
+    public Class<?> type() {
+        return TaskAttemptExecutionAPIEntity.class;
+    }
+
+    @Override
+    public void flush(TaskAttemptExecutionAPIEntity entity) {
+        Map<String, Object> fields = new HashMap<>(entity.getTags());
+        fields.put("startTime", entity.getStartTime());
+        fields.put("endTime", entity.getEndTime());
+        fields.put("taskStatus", entity.getTaskStatus());
+        if (!fields.containsKey(MRJobTagName.ERROR_CATEGORY.toString())) {
+            fields.put("errorCategory", "");
+        }
+        collector.collect(stormStreamId, new ValuesArray(fields.get(MRJobTagName.TASK_ATTEMPT_ID.toString()),
fields));
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8fe968cb/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/storm/JobHistorySpout.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/storm/JobHistorySpout.java
b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/storm/JobHistorySpout.java
index 9743ea7..2157da9 100644
--- a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/storm/JobHistorySpout.java
+++ b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/storm/JobHistorySpout.java
@@ -97,10 +97,7 @@ public class JobHistorySpout extends BaseRichSpout {
     private static final int MAX_RETRY_TIMES = 3;
     private MRHistoryJobConfig appConfig;
     private JobHistoryEndpointConfig jobHistoryEndpointConfig;
-
-    public JobHistorySpout(JobHistoryContentFilter filter, MRHistoryJobConfig appConfig)
{
-        this(filter, appConfig, new JobHistorySpoutCollectorInterceptor());
-    }
+    private List<String> streams;
 
     /**
      * mostly this constructor signature is for unit test purpose as you can put customized
interceptor here.
@@ -178,12 +175,20 @@ public class JobHistorySpout extends BaseRichSpout {
         }
     }
 
+    public void setStreams(List<String> streams) {
+        this.streams = streams;
+    }
+
     /**
      * empty because framework will take care of output fields declaration.
      */
     @Override
     public void declareOutputFields(OutputFieldsDeclarer declarer) {
-        declarer.declare(new Fields("jobId", "message"));
+        if (streams != null) {
+            for (String streamId : streams) {
+                declarer.declareStream(streamId, new Fields("f1", "message"));
+            }
+        }
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8fe968cb/eagle-jpm/eagle-jpm-mr-history/src/main/resources/META-INF/providers/org.apache.eagle.jpm.mr.history.MRHistoryJobApplicationProvider.xml
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-mr-history/src/main/resources/META-INF/providers/org.apache.eagle.jpm.mr.history.MRHistoryJobApplicationProvider.xml
b/eagle-jpm/eagle-jpm-mr-history/src/main/resources/META-INF/providers/org.apache.eagle.jpm.mr.history.MRHistoryJobApplicationProvider.xml
index 835dbda..1c3b5cb 100644
--- a/eagle-jpm/eagle-jpm-mr-history/src/main/resources/META-INF/providers/org.apache.eagle.jpm.mr.history.MRHistoryJobApplicationProvider.xml
+++ b/eagle-jpm/eagle-jpm-mr-history/src/main/resources/META-INF/providers/org.apache.eagle.jpm.mr.history.MRHistoryJobApplicationProvider.xml
@@ -35,8 +35,14 @@
             <value>4</value>
         </property>
         <property>
-            <name>stormConfig.historyKafkaSinkTasks</name>
-            <displayName>Sink Task Number</displayName>
+            <name>stormConfig.jobKafkaSinkTasks</name>
+            <displayName>Sink Task Number For Job Stream</displayName>
+            <description>the number tasks of the sink bolt will be assigned</description>
+            <value>1</value>
+        </property>
+        <property>
+            <name>stormConfig.taskAttemptKafkaSinkTasks</name>
+            <displayName>Sink Task Number For Task Attempt Stream</displayName>
             <description>the number tasks of the sink bolt will be assigned</description>
             <value>1</value>
         </property>
@@ -144,8 +150,8 @@
     </configuration>
     <streams>
         <stream>
-            <streamId>map_reduce_failed_job_stream</streamId>
-            <description>Map Reduce Failed Job Stream</description>
+            <streamId>MAP_REDUCE_JOB_STREAM</streamId>
+            <description>Map Reduce Job Stream</description>
             <validate>true</validate>
             <columns>
                 <column>
@@ -169,10 +175,6 @@
                     <type>string</type>
                 </column>
                 <column>
-                    <name>jobName</name>
-                    <type>string</type>
-                </column>
-                <column>
                     <name>jobId</name>
                     <type>string</type>
                 </column>
@@ -198,6 +200,69 @@
                 </column>
             </columns>
         </stream>
+        <stream>
+            <streamId>MAP_REDUCE_TASK_ATTEMPT_STREAM</streamId>
+            <description>Map Reduce Task Attempt Stream</description>
+            <validate>true</validate>
+            <columns>
+                <column>
+                    <name>site</name>
+                    <type>string</type>
+                </column>
+                <column>
+                    <name>rack</name>
+                    <type>string</type>
+                </column>
+                <column>
+                    <name>queue</name>
+                    <type>string</type>
+                </column>
+                <column>
+                    <name>user</name>
+                    <type>string</type>
+                </column>
+                <column>
+                    <name>jobType</name>
+                    <type>string</type>
+                </column>
+                <column>
+                    <name>jobDefId</name>
+                    <type>string</type>
+                </column>
+                <column>
+                    <name>jobId</name>
+                    <type>string</type>
+                </column>
+                <column>
+                    <name>taskId</name>
+                    <type>string</type>
+                </column>
+                <column>
+                    <name>taskType</name>
+                    <type>string</type>
+                </column>
+                <column>
+                    <name>taskAttemptId</name>
+                    <type>string</type>
+                </column>
+                <column>
+                    <name>errorCategory</name>
+                    <type>string</type>
+                </column>
+                <column>
+                    <name>startTime</name>
+                    <type>long</type>
+                </column>
+                <column>
+                    <name>endTime</name>
+                    <type>long</type>
+                </column>
+                <column>
+                    <name>taskStatus</name>
+                    <type>string</type>
+                </column>
+            </columns>
+        </stream>
     </streams>
     <docs>
         <install>

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8fe968cb/eagle-jpm/eagle-jpm-mr-history/src/main/resources/application.conf
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-mr-history/src/main/resources/application.conf b/eagle-jpm/eagle-jpm-mr-history/src/main/resources/application.conf
index 4d4dc4b..e8d5311 100644
--- a/eagle-jpm/eagle-jpm-mr-history/src/main/resources/application.conf
+++ b/eagle-jpm/eagle-jpm-mr-history/src/main/resources/application.conf
@@ -22,7 +22,8 @@
 
   "stormConfig" : {
     "mrHistoryJobSpoutTasks" : 6,
-    "historyKafkaSinkTasks" : 1
+    "jobKafkaSinkTasks" : 1,
+    "taskAttemptKafkaSinkTasks" : 1
   },
 
   "zookeeper" : {



Mime
View raw message