Learning Storm Packt 2014 [推广有奖]

11楼

Lisrelchen 发表于 2016-4-17 02:50:21 |只看作者 |坛友微信交流群

package com.learningstorm.monitoring;
import java.util.Iterator;
import java.util.Map;
import backtype.storm.generated.ExecutorSpecificStats;
import backtype.storm.generated.ExecutorStats;
import backtype.storm.generated.ExecutorSummary;
import backtype.storm.generated.Nimbus.Client;
import backtype.storm.generated.SpoutStats;
import backtype.storm.generated.TopologyInfo;
public class SpoutStatistics {
private static final String DEFAULT = "default";
private static final String ALL_TIME = ":all-time";
public void printSpoutStatistics(String topologyId) {
try {
ThriftClient thriftClient = new ThriftClient();
// Get the nimbus thrift client
Client client = thriftClient.getClient();
// Get the information of given topology
TopologyInfo topologyInfo = client.getTopologyInfo(topologyId);
Iterator<ExecutorSummary> executorSummaryIterator = topologyInfo
.get_executors_iterator();
while (executorSummaryIterator.hasNext()) {
ExecutorSummary executorSummary = executorSummaryIterator.next();
ExecutorStats executorStats = executorSummary.get_stats();
if(executorStats !=null) {
ExecutorSpecificStats executorSpecificStats = executorStats.get_specific();
String componentId = executorSummary.get_component_id();
//
if (executorSpecificStats.is_set_spout()) {
SpoutStats spoutStats = executorSpecificStats.get_spout();
System.out.println("*************************************");
System.out.println("Component ID of Spout:- " + componentId);
System.out.println("Transferred:- "
+ getAllTimeStat(executorStats.get_transferred(),ALL_TIME));
System.out.println("Total tuples emitted:- "
+ getAllTimeStat(executorStats.get_emitted(), ALL_TIME));
System.out.println("Acked: "
+ getAllTimeStat(spoutStats.get_acked(),
ALL_TIME));
System.out.println("Failed: "
+ getAllTimeStat(spoutStats.get_failed(),
ALL_TIME));
System.out.println("*************************************");
}
}
}
}catch (Exception exception) {
throw new RuntimeException("Error occurred while fetching the spout information : "+exception);
}
}
private static Long getAllTimeStat(Map<String, Map<String, Long>> map,
String statName) {
if (map != null) {
Long statValue = null;
Map<String, Long> tempMap = map.get(statName);
statValue = tempMap.get(DEFAULT);
return statValue;
}
return 0L;
}
public static void main(String[] args) {
new SpoutStatistics().printSpoutStatistics("LearningStormClusterTopology-1-1393847956");
}
}

复制代码

使用道具举报

12楼

Lisrelchen 发表于 2016-4-17 02:51:43 |只看作者 |坛友微信交流群

package com.learningstorm.monitoring;
import java.util.Iterator;
import backtype.storm.generated.ClusterSummary;
import backtype.storm.generated.Nimbus.Client;
import backtype.storm.generated.SupervisorSummary;
public class SupervisorStatistics {
public void printSupervisorStatistics() {
try {
ThriftClient thriftClient = new ThriftClient();
Client client = thriftClient.getClient();
// Get the cluster information.
ClusterSummary clusterSummary = client.getClusterInfo();
// Get the SupervisorSummary interator
Iterator<SupervisorSummary> supervisorsIterator = clusterSummary.get_supervisors_iterator();
while (supervisorsIterator.hasNext()) {
// Print the information of supervisor node
SupervisorSummary supervisorSummary = (SupervisorSummary) supervisorsIterator.next();
System.out.println("*************************************");
System.out.println("Supervisor Host IP : "+supervisorSummary.get_host());
System.out.println("Number of used workers : "+supervisorSummary.get_num_used_workers());
System.out.println("Number of workers : "+supervisorSummary.get_num_workers());
System.out.println("Supervisor ID : "+supervisorSummary.get_supervisor_id());
System.out.println("Supervisor uptime in seconds : "+supervisorSummary.get_uptime_secs());
System.out.println("*************************************");
}
}catch (Exception e) {
throw new RuntimeException("Error occurred while getting cluster info : ");
}
}
public static void main(String[] args) {
new SupervisorStatistics().printSupervisorStatistics();
}
}