Hands-On Industrial Internet of Things

Implementing our analytics

We are now ready to run our analytics. On the <KAIROSDB_HOME>/plugins directory, we need to create a Python file called mymean_analytic.py. The code for this is as follows:

from datetime import datetimefrom airflow import DAGfrom airflow.operators.dummy_operator import DummyOperatorfrom airflow.operators.python_operator import PythonOperatorfrom airflow.operators import KairosDBOperatorimport datetimeimport loggingdef my_mean(ds, **kwargs):    logging.info("kwargs: %s" % kwargs)    logging.info("ds: %s" % ds)    ti = kwargs['ti']    data = ti.xcom_pull(key=None, task_ids=['get_data'])    return _mean(data)def _mean(data):    ret={}    for d in data:        results = d['results']        for r in results: m = [float(sum(l))/len(l) for l in zip(*r['values'])] ...