Block Code Example to generate Metrics
This sample Block code generates some grouped Bar Graph and publishes to the metric server so that it can be visualized in the Pipeline run results interface under 'Metrics' tab.
from pipelineblocksdk.api.Singleton import Singleton
from pipelineblocksdk.construct.base.BatchBlock import BatchBlock
# Following code is the definition for a batch block
class MyBlock(BatchBlock, Singleton):
# This is the entrypoint for the block. This is a mandatory function.
def run(self):
# -------- Generating Bar Graphs -------------
# Metric default value
self.metrics["keys"] = self.block_params
self.metrics["name"] = "BAR_GRAPH_V1.0"
self.logger.info("Generating graphs....")
self.metrics["metricId"] = "Animals In Zoo"
self.logger.info("Creating graph for : " + "Animals In Zoo")
self.metrics["metrics"] = {
str(int(1)): {
"x": "Giraffes",
"y": [20,12],
"x_title": "Animals",
"y_title": "No : of Animals",
"y_label": ["SF Zoo", "LA Zoo"],
"colors": ["blue", "rgba(255,0,0,.8)"],
"chart_type": "group"
}
}
self.logger.info("Sending data to metric server... ")
self.metrics_handler.save_metrics_time(self.metrics)
self.metrics["metrics"] = {
str(int(2)): {
"x": "Monkey",
"y": [30,8],
"x_title": "Animals",
"y_title": "No : of Animals",
"y_label": ["SF Zoo", "LA Zoo"],
"colors": ["blue", "rgba(255,0,0,.8)"],
"chart_type": "group"
}
}
self.logger.info("Sending data to metric server... ")
self.metrics_handler.save_metrics_time(self.metrics)
self.metrics["metrics"] = {
str(int(3)): {
"x": "Deer",
"y": [34,19],
"x_title": "Animals",
"y_title": "No : of Animals",
"y_label": ["SF Zoo", "LA Zoo"],
"colors": ["blue", "rgba(255,0,0,.8)"],
"chart_type": "group"
}
}
self.logger.info("Sending data to metric server... ")
self.metrics_handler.save_metrics_time(self.metrics)
# -------- Generating Line Graphs -------------
# Metric default value
self.metrics["keys"] = self.block_params
self.metrics["name"] = "LINE_GRAPH_v1.1"
self.logger.info("Generating graphs....")
self.metrics["appId"] = "METRIC_LINE_GRAPH"
self.metrics["metricId"] = "No : of Giraffes over 3 days in a Zoo"
self.logger.info("Creating graph for : " + "No : of Giraffes over 3 days in a Zoo")
self.metrics["metrics"] = {
str(int(time.time()*1000000)): {
"x": 1,
"y": [20,12],
"x_label": "Days",
"y_label": ["SF Zoo", "LA Zoo"],
"x_type": "INTEGER",
"y_type": "INTEGER"
}
}
self.logger.info("Sending data to metric server... ")
self.metrics_handler.save_metrics_time(self.metrics)
self.metrics["metrics"] = {
str(int(time.time()*1000000)): {
"x":2,
"y": [30,8],
"x_label": "Zoo",
"y_label": ["SF Zoo", "LA Zoo"],
"x_type": "INTEGER",
"y_type": "INTEGER"
}
}
self.logger.info("Sending data to metric server... ")
self.metrics_handler.save_metrics_time(self.metrics)
self.metrics["metrics"] = {
str(int(time.time()*1000000)): {
"x":3,
"y": [35,18],
"x_label": "Zoo",
"y_label": ["SF Zoo", "LA Zoo"],
"x_type": "INTEGER",
"y_type": "INTEGER"
}
}
self.logger.info("Sending data to metric server... ")
self.metrics_handler.save_metrics_time(self.metrics)
# Set the output parameter
output_dict = dict()
# Set the status of the block as completed
self.block_status = "COMPLETED"
return output_dict