EdgeAI:   Google Coral with Coral Environmental Sensors and TPU With NiFi and MiNiFi (Updated EFM)
Tags: 2020, agent, ai, apachenifi
Published: 2020-02-22 20:27:40

---
title: EdgeAI:   Google Coral with Coral Environmental Sensors and TPU With NiFi and MiNiFi (Updated EFM)
published: true
date: 2020-02-21 21:08:00 UTC
tags: 2020,agent,AI,apache-nifi
canonical_url: 
---

## EdgeAI:   Google Coral with Coral Environmental Sensors and TPU With NiFi and MiNiFi

  

**Building MiNiFi IoT Apps with the new Cloudera EFM **

  

[![](https://1.bp.blogspot.com/-fLY86Ijv0Iw/XlAz-t-AEcI/AAAAAAAAZeY/CA28HxW8Z8Eo5fhszVkNG3sfOu1LcYH3ACLcBGAsYHQ/s640/efmCoralFlow.png)](https://1.bp.blogspot.com/-fLY86Ijv0Iw/XlAz-t-AEcI/AAAAAAAAZeY/CA28HxW8Z8Eo5fhszVkNG3sfOu1LcYH3ACLcBGAsYHQ/s1600/efmCoralFlow.png)

  

It is very easy to build a drag and drop EdgeAI application with EFM and then push to all your MiNiFi agents.

  

  

Cloudera Edge Management CEM-1.1.1  
Download the newest CEM today!

[https://www.cloudera.com/downloads/cdf/cem.html](https://www.cloudera.com/downloads/cdf/cem.html)

[https://docs.cloudera.com/cem/1.1.1/release-notes/topics/cem-whats-new.html](https://docs.cloudera.com/cem/1.1.1/release-notes/topics/cem-whats-new.html)

  

[![](https://1.bp.blogspot.com/-6EwCP0JGJmI/XlA__FNuDgI/AAAAAAAAZfg/kiXOZNaWagsdObA0gIXQEzA5naTDEcNewCLcBGAsYHQ/s640/cemMonitorServer.png)](https://1.bp.blogspot.com/-6EwCP0JGJmI/XlA__FNuDgI/AAAAAAAAZfg/kiXOZNaWagsdObA0gIXQEzA5naTDEcNewCLcBGAsYHQ/s1600/cemMonitorServer.png)
  

[![](https://1.bp.blogspot.com/-4uB14DjkCzU/XlA__Fbq1zI/AAAAAAAAZfc/l9ctQmVkMdYe3sxPij8umjQoaW_ddod_wCLcBGAsYHQ/s640/efmAgentClassMonitoring.png)](https://1.bp.blogspot.com/-4uB14DjkCzU/XlA__Fbq1zI/AAAAAAAAZfc/l9ctQmVkMdYe3sxPij8umjQoaW_ddod_wCLcBGAsYHQ/s1600/efmAgentClassMonitoring.png)
  

[![](https://1.bp.blogspot.com/--HnM4ZyyNOU/XlA__B3lQFI/AAAAAAAAZfY/cXlzkDKNuCsA3kRB4DKm7kKPycVti7fgwCLcBGAsYHQ/s640/efmAgentInfo.png)](https://1.bp.blogspot.com/--HnM4ZyyNOU/XlA__B3lQFI/AAAAAAAAZfY/cXlzkDKNuCsA3kRB4DKm7kKPycVti7fgwCLcBGAsYHQ/s1600/efmAgentInfo.png)
  

[![](https://1.bp.blogspot.com/-sYA33aP4vVU/XlBAAGXnXlI/AAAAAAAAZfs/J9LEWOWV9ZYVKOb8IbtKhw45bjjME317ACLcBGAsYHQ/s640/efmclassinfo.png)](https://1.bp.blogspot.com/-sYA33aP4vVU/XlBAAGXnXlI/AAAAAAAAZfs/J9LEWOWV9ZYVKOb8IbtKhw45bjjME317ACLcBGAsYHQ/s1600/efmclassinfo.png)
  

[![](https://1.bp.blogspot.com/-GHN_h4BnKQM/XlA__5Qc80I/AAAAAAAAZfk/wxyr5_q0Nxw5-e3gRkr3vz8MUcIqCFDcwCLcBGAsYHQ/s640/efmDashboard.png)](https://1.bp.blogspot.com/-GHN_h4BnKQM/XlA__5Qc80I/AAAAAAAAZfk/wxyr5_q0Nxw5-e3gRkr3vz8MUcIqCFDcwCLcBGAsYHQ/s1600/efmDashboard.png)
  

[![](https://1.bp.blogspot.com/-OqLplj9RnWE/XlBAAMsu23I/AAAAAAAAZfo/lzElKIADx8ERKtH-OJVuf86lKeVoWThmQCLcBGAsYHQ/s640/efmParameters.png)](https://1.bp.blogspot.com/-OqLplj9RnWE/XlBAAMsu23I/AAAAAAAAZfo/lzElKIADx8ERKtH-OJVuf86lKeVoWThmQCLcBGAsYHQ/s1600/efmParameters.png)

  

  

  

**NiFi Flow Receiving From MiNiFi Java Agent**

  

[![](https://1.bp.blogspot.com/-jEVdFoDBveU/XlAusbzUzsI/AAAAAAAAZdo/b7OxmtM0h_sc42nCOXWnuIshxtEYfsSUwCEwYBhgL/s640/coralInitialFlowIngest.png)](https://1.bp.blogspot.com/-jEVdFoDBveU/XlAusbzUzsI/AAAAAAAAZdo/b7OxmtM0h_sc42nCOXWnuIshxtEYfsSUwCEwYBhgL/s1600/coralInitialFlowIngest.png)

  

In a cluster in my CDP-DC Cluster I consume Kafka messages sent from my remote **NiFi** gateway to publish alerts to **Kafka** and push records to Apache **HBase** and Apache **Kudu**.  We filter our data with Streaming SQL.
  

[![](https://1.bp.blogspot.com/-TLX3-auY6Ko/XlAus9oRmjI/AAAAAAAAZdk/YCl7MhgSTms0bZeNddbirsj3-80MwQXVACEwYBhgL/s640/coralSensorsToHbaseAndKudu.png)](https://1.bp.blogspot.com/-TLX3-auY6Ko/XlAus9oRmjI/AAAAAAAAZdk/YCl7MhgSTms0bZeNddbirsj3-80MwQXVACEwYBhgL/s1600/coralSensorsToHbaseAndKudu.png)

  

**We can use SQL to route, create aggregates like averages, chose a subset of fields and limit data returned.   Using the power of Apache Calcite, Streaming SQL in NiFi is a game changer against Record Data Types including CSV, XML, Avro, Parquet, JSON and Grokable text.   Read and write different formats and convert when your SQL is done.   Or just to SELECT \* FROM FLOWFILE to get everything.  **

  

[![](https://1.bp.blogspot.com/-vWhFIMdfA2Y/XlA2JgtH6qI/AAAAAAAAZe8/nQt_N_BRcu0fDRswom3qj93l3xOjDwsZACLcBGAsYHQ/s640/sqlDetailsCoral.png)](https://1.bp.blogspot.com/-vWhFIMdfA2Y/XlA2JgtH6qI/AAAAAAAAZe8/nQt_N_BRcu0fDRswom3qj93l3xOjDwsZACLcBGAsYHQ/s1600/sqlDetailsCoral.png)
  
  
We can see this flow from **Atlas** as we trace the data lineage and provenance from **Kafka** topic.  
  

[![](https://1.bp.blogspot.com/-uA_NHspAThg/XlAur17LzjI/AAAAAAAAZdU/g-Db6fzyqu4rDb_pXsnDZawzIuzXGLV-ACEwYBhgL/s640/coral%2540Princeton0.png)](https://1.bp.blogspot.com/-uA_NHspAThg/XlAur17LzjI/AAAAAAAAZdU/g-Db6fzyqu4rDb_pXsnDZawzIuzXGLV-ACEwYBhgL/s1600/coral%2540Princeton0.png)
  
  

**We can search Atlas for Kafka Topics.**
  

[![](https://1.bp.blogspot.com/-_nsEIBi0_WI/XlAuqa15rcI/AAAAAAAAZdQ/AsJqU2savBQ_yMsquaXhqXPWWz67rmkUwCEwYBhgL/s640/atlasKafkaTopicSearch.png)](https://1.bp.blogspot.com/-_nsEIBi0_WI/XlAuqa15rcI/AAAAAAAAZdQ/AsJqU2savBQ_yMsquaXhqXPWWz67rmkUwCEwYBhgL/s1600/atlasKafkaTopicSearch.png)
  
  

**From coral Kafka topic to NiFi to Kudu.**
  

[![](https://1.bp.blogspot.com/-4AMYt3v_vSk/XlAupjiPtZI/AAAAAAAAZdg/B-oEeePMbCwuAvP8v3sHbZGVfgOwrx4vgCEwYBhgL/s640/89d039dd-6a5a-1e18-bb78-321dc894022e%2540Princeton0.png)](https://1.bp.blogspot.com/-4AMYt3v_vSk/XlAupjiPtZI/AAAAAAAAZdg/B-oEeePMbCwuAvP8v3sHbZGVfgOwrx4vgCEwYBhgL/s1600/89d039dd-6a5a-1e18-bb78-321dc894022e%2540Princeton0.png)

  

**Details on Coral Kafka Topic**

[![](https://1.bp.blogspot.com/-nbtjKTLTR6E/XlA2IOCFdVI/AAAAAAAAZek/kuIUO8_dX9cwCn-y1X6ysfjb9ghTZUXOQCLcBGAsYHQ/s640/atlasBrowseCoralTopic.png)](https://1.bp.blogspot.com/-nbtjKTLTR6E/XlA2IOCFdVI/AAAAAAAAZek/kuIUO8_dX9cwCn-y1X6ysfjb9ghTZUXOQCLcBGAsYHQ/s1600/atlasBrowseCoralTopic.png)

  

  

**Examining the Hive Metastore Data on the Coral Kudu Table**
  

[![](https://1.bp.blogspot.com/-YDqhcnbRYso/XlA2IJBZHHI/AAAAAAAAZes/CwzW0crVruEC1KJYiKaDQT6zK_vCzQ53gCLcBGAsYHQ/s640/atlasCoralTable.png)](https://1.bp.blogspot.com/-YDqhcnbRYso/XlA2IJBZHHI/AAAAAAAAZes/CwzW0crVruEC1KJYiKaDQT6zK_vCzQ53gCLcBGAsYHQ/s1600/atlasCoralTable.png)

  

**NiFi Flow Details in Atlas**
  

[![](https://1.bp.blogspot.com/-fLI6p60JXHU/XlA2IEkprWI/AAAAAAAAZeo/G5AFOmzPbdcQQRh96WcQpOYOgI8wGmBMwCLcBGAsYHQ/s640/atlasNiFiFlowDetails.png)](https://1.bp.blogspot.com/-fLI6p60JXHU/XlA2IEkprWI/AAAAAAAAZeo/G5AFOmzPbdcQQRh96WcQpOYOgI8wGmBMwCLcBGAsYHQ/s1600/atlasNiFiFlowDetails.png)
  

**Details on Alerts Topic**

[![](https://1.bp.blogspot.com/-7hIJHFoVkaM/XlA2I4208sI/AAAAAAAAZew/si-MM4hD7FECL94bqKwzp6REOPbpoOgZACLcBGAsYHQ/s640/atlasSendAlert.png)](https://1.bp.blogspot.com/-7hIJHFoVkaM/XlA2I4208sI/AAAAAAAAZew/si-MM4hD7FECL94bqKwzp6REOPbpoOgZACLcBGAsYHQ/s1600/atlasSendAlert.png)'

  

  

**Statistics from Atlas**
  

[![](https://1.bp.blogspot.com/-DTEQvbZ8fEI/XlA2JVZTuvI/AAAAAAAAZe4/QtLYq7bFbIEE5mst5sJzzkVBbp1fofSkACLcBGAsYHQ/s640/atlasStats2.png)](https://1.bp.blogspot.com/-DTEQvbZ8fEI/XlA2JVZTuvI/AAAAAAAAZe4/QtLYq7bFbIEE5mst5sJzzkVBbp1fofSkACLcBGAsYHQ/s1600/atlasStats2.png)
  

[![](https://1.bp.blogspot.com/-DVpJF8bBxqQ/XlA2JH3AK6I/AAAAAAAAZe0/1ClM6TbrM-MWQfdMCanESg6H5eFLFERNQCLcBGAsYHQ/s640/atlasStats.png)](https://1.bp.blogspot.com/-DVpJF8bBxqQ/XlA2JH3AK6I/AAAAAAAAZe0/1ClM6TbrM-MWQfdMCanESg6H5eFLFERNQCLcBGAsYHQ/s1600/atlasStats.png)
  

**See:   ** [https://www.datainmotion.dev/2020/02/connecting-apache-nifi-to-apache-atlas.html](https://dev.to/tspannhw/connecting-apache-nifi-to-apache-atlas-for-data-governance-at-scale-in-streaming-1p09-temp-slug-1697049)

  

  

**Example Web Camera Image**
  

[![](https://1.bp.blogspot.com/-dCdPuxphhPA/XlAupfHDx4I/AAAAAAAAZdk/pEVM1EmvO9AvQ5OtYonHNYvTO_yISe7aQCEwYBhgL/s640/2020-02-20_170816.jpg)](https://1.bp.blogspot.com/-dCdPuxphhPA/XlAupfHDx4I/AAAAAAAAZdk/pEVM1EmvO9AvQ5OtYonHNYvTO_yISe7aQCEwYBhgL/s1600/2020-02-20_170816.jpg)
  

  

** Example JSON Record**
  

> [{"cputemp":59,"id":"20200221190718\_2632409e-f635-48e7-9f32-aa1333f3b8f9","temperature":"39.44","memory":91.1,"score\_1":"0.29","starttime":"02/21/2020 14:07:13","label\_1":"hair spray","tempf":"102.34","diskusage":"50373.5 MB","message":"Success","ambient\_light":"329.92","host":"coralenv","cpu":34.1,"macaddress":"b8:27:eb:99:64:6b","pressure":"102.76","score\_2":"0.14","ip":"127.0.1.1","te":"5.10","systemtime":"02/21/2020 14:07:18","label\_2":"syringe","humidity":"10.21"}]

  
  

**Querying Kudu results in Hue**

  

[![](https://1.bp.blogspot.com/-7sBmR9BLwl4/XlAusH9OuhI/AAAAAAAAZdU/a3TeibQZjG0C9Sd7L-XR40NIUyqE2oUXwCEwYBhgL/s640/coralHueData.png)](https://1.bp.blogspot.com/-7sBmR9BLwl4/XlAusH9OuhI/AAAAAAAAZdU/a3TeibQZjG0C9Sd7L-XR40NIUyqE2oUXwCEwYBhgL/s1600/coralHueData.png)

  

**Pushing Alerts to Slack from NiFi**
  

[![](https://1.bp.blogspot.com/-DiMjyOvVxZ0/XlAutJXZ9-I/AAAAAAAAZdk/KNTQJvN-XoMAst6wy8EMfd5xtGmLdfVBgCEwYBhgL/s640/coralSlackResults.png)](https://1.bp.blogspot.com/-DiMjyOvVxZ0/XlAutJXZ9-I/AAAAAAAAZdk/KNTQJvN-XoMAst6wy8EMfd5xtGmLdfVBgCEwYBhgL/s1600/coralSlackResults.png)
  

  

  

  

**I am running on Apache NiFi 1.11.1 and wanted to point out a new feature.   Download flow:   Will download the highlighted flow/pgroup as JSON.**
  

[![](https://1.bp.blogspot.com/-9VwvGvE4WTU/XlAuusRNSNI/AAAAAAAAZdk/J2rJNuIKCIEifAA2w0qpUaRLRJ1nQ5WEACEwYBhgL/s320/nifi11downloadFlowAsJSON.png)](https://1.bp.blogspot.com/-9VwvGvE4WTU/XlAuusRNSNI/AAAAAAAAZdk/J2rJNuIKCIEifAA2w0qpUaRLRJ1nQ5WEACEwYBhgL/s1600/nifi11downloadFlowAsJSON.png)
  

  

  
**Looking at NiFi counters to monitor progress:**  
  

[![](https://1.bp.blogspot.com/-7F3Es0V8OSY/XlAykkYU4GI/AAAAAAAAZeE/_tqukAsj9NorXAo4RfjMMEyvDipsCEZ_gCLcBGAsYHQ/s1600/nif1.11.counters.png)](https://1.bp.blogspot.com/-7F3Es0V8OSY/XlAykkYU4GI/AAAAAAAAZeE/_tqukAsj9NorXAo4RfjMMEyvDipsCEZ_gCLcBGAsYHQ/s1600/nif1.11.counters.png)

### We can see how easy it is to ingest IoT sensor data and run AI algorithms on Coral TPUs.

  

  

**Shell (coralrun.sh)**

  

<style type="text/css">p.p1 {margin: 0.0px 0.0px 0.0px 0.0px; font: 18.0px Menlo; color: #ffffff; background-color: #224fbc} span.s1 {font-variant-ligatures: no-common-ligatures} </style>
#!/bin/bash

DATE=$(date +"%Y-%m-%d\_%H%M%S")

fswebcam -q -r 1280x720 /opt/demo/images/$DATE.jpg

python3 -W ignore /opt/demo/test.py --image /opt/demo/images/$DATE.jpg 2>/dev/null

  
  
**Kudu Table DDL**  
  
[https://github.com/tspannhw/table-ddl](https://github.com/tspannhw/table-ddl)  
  
  
**Python 3 (test.py)**  
  

> <style type="text/css">p.p1 {margin: 0.0px 0.0px 0.0px 0.0px; font: 18.0px Menlo; color: #ffffff; background-color: #224fbc} p.p2 {margin: 0.0px 0.0px 0.0px 0.0px; font: 18.0px Menlo; color: #ffffff; background-color: #224fbc; min-height: 21.0px} span.s1 {font-variant-ligatures: no-common-ligatures} </style>
> import time
> 
> import sys
> 
> import subprocess
> 
> import os
> 
> import base64
> 
> import uuid
> 
> import datetime
> 
> import traceback
> 
> import base64
> 
> import json
> 
> from time import gmtime, strftime
> 
> import math
> 
> import random, string
> 
> import time
> 
> import psutil
> 
> import uuid 
> 
> from getmac import get\_mac\_address
> 
> from coral.enviro.board import EnviroBoard
> 
> from luma.core.render import canvas
> 
> from PIL import Image, ImageDraw, ImageFont
> 
> import os
> 
> import argparse
> 
> from edgetpu.classification.engine import ClassificationEngine
> 
>   
> 
> # Importing socket library 
> 
> import socket 
> 
>   
> 
> start = time.time()
> 
> starttf = datetime.datetime.now().strftime('%m/%d/%Y %H:%M:%S')
> 
>   
> 
> def ReadLabelFile(file\_path):
> 
>     with open(file\_path, 'r') as f:
> 
>         lines = f.readlines()
> 
>     ret = {}
> 
>     for line in lines:
> 
>         pair = line.strip().split(maxsplit=1)
> 
>         ret[int(pair[0])] = pair[1].strip()
> 
>     return ret
> 
>   
> 
> # Google Example Code
> 
> def update\_display(display, msg):
> 
>     with canvas(display) as draw:
> 
>         draw.text((0, 0), msg, fill='white')
> 
>   
> 
> def getCPUtemperature():
> 
>     res = os.popen('vcgencmd measure\_temp').readline()
> 
>     return(res.replace("temp=","").replace("'C\n",""))
> 
>   
> 
> # Get MAC address of a local interfaces
> 
> def psutil\_iface(iface):
> 
>     # type: (str) -> Optional[str]
> 
>     import psutil
> 
>     nics = psutil.net\_if\_addrs()
> 
>     if iface in nics:
> 
>         nic = nics[iface]
> 
>         for i in nic:
> 
>             if i.family == psutil.AF\_LINK:
> 
>                 return i.address
> 
> # /opt/demo/examples-camera/all\_models  
> 
> row = { }
> 
> try:
> 
> #i = 1
> 
> #while i == 1:
> 
>     parser = argparse.ArgumentParser()
> 
>     parser.add\_argument('--image', help='File path of the image to be recognized.', required=True)
> 
>     args = parser.parse\_args()
> 
>     # Prepare labels.
> 
>     labels = ReadLabelFile('/opt/demo/examples-camera/all\_models/imagenet\_labels.txt')
> 
>   
> 
>     # Initialize engine.
> 
>     engine = ClassificationEngine('/opt/demo/examples-camera/all\_models/inception\_v4\_299\_quant\_edgetpu.tflite')
> 
>   
> 
>     # Run inference.
> 
>     img = Image.open(args.image)
> 
>   
> 
>     scores = {}
> 
>     kCount = 1
> 
>   
> 
>     # Iterate Inference Results
> 
>     for result in engine.ClassifyWithImage(img, top\_k=5):
> 
>         scores['label\_' + str(kCount)] = labels[result[0]]
> 
>         scores['score\_' + str(kCount)] = "{:.2f}".format(result[1])
> 
>         kCount = kCount + 1    
> 
>   
> 
>     enviro = EnviroBoard()
> 
>     host\_name = socket.gethostname()
> 
>     host\_ip = socket.gethostbyname(host\_name)
> 
>     cpuTemp=int(float(getCPUtemperature()))
> 
>     uuid2 = '{0}\_{1}'.format(strftime("%Y%m%d%H%M%S",gmtime()),uuid.uuid4())
> 
>     usage = psutil.disk\_usage("/")
> 
>     end = time.time()
> 
>     row.update(scores)
> 
>     row['host'] = os.uname()[1]
> 
>     row['ip'] = host\_ip
> 
>     row['macaddress'] = psutil\_iface('wlan0')
> 
>     row['cputemp'] = round(cpuTemp,2)
> 
>     row['te'] = "{0:.2f}".format((end-start))
> 
>     row['starttime'] = starttf
> 
>     row['systemtime'] = datetime.datetime.now().strftime('%m/%d/%Y %H:%M:%S')
> 
>     row['cpu'] = psutil.cpu\_percent(interval=1)
> 
>     row['diskusage'] = "{:.1f} MB".format(float(usage.free) / 1024 / 1024)
> 
>     row['memory'] = psutil.virtual\_memory().percent
> 
>     row['id'] = str(uuid2)
> 
>     row['message'] = "Success"
> 
>     row['temperature'] = '{0:.2f}'.format(enviro.temperature)
> 
>     row['humidity'] = '{0:.2f}'.format(enviro.humidity)
> 
>     row['tempf'] = '{0:.2f}'.format((enviro.temperature \* 1.8) + 32)    
> 
>     row['ambient\_light'] = '{0}'.format(enviro.ambient\_light)
> 
>     row['pressure'] = '{0:.2f}'.format(enviro.pressure)
> 
>     msg = 'Temp: {0}'.format(row['temperature'])
> 
>     msg += 'IP: {0}'.format(row['ip'])
> 
>     update\_display(enviro.display, msg)
> 
> #    i = 2
> 
> except:
> 
>     row['message'] = "Error"
> 
> print(json.dumps(row)) 

###   

### **Source Code:**

  

[https://github.com/tspannhw/nifi-minifi-coral-env](https://github.com/tspannhw/nifi-minifi-coral-env)

  

### Sensors / Devices / Hardware:

- Humdity-HDC2010 humidity sensor
- Light-OPT3002 ambient light sensor
- Barometric-BMP280 barometric pressure sensor
- PS3 Eye Camera and Microphone USB
- Raspberry Pi 3B+
- Google Coral Environmental Sensor Board
- Google Coral USB Accelerator TPU

  

### References:

  

- [https://coral.ai/docs/enviro-board/get-started/](https://coral.ai/docs/enviro-board/get-started/)
- [https://coral.ai/products/accelerator/](https://coral.ai/products/accelerator/)
- [https://coral.ai/docs/enviro-board/datasheet/](https://coral.ai/docs/enviro-board/datasheet/)
- [https://github.com/tspannhw/nifi-minifi-coral-env](https://github.com/tspannhw/nifi-minifi-coral-env)
- [https://github.com/tspannhw/nifi-minifi-coral](https://github.com/tspannhw/nifi-minifi-coral)
- [https://www.datainmotion.dev/2019/08/google-coral-tpu-with-edge-devices-and.html](https://dev.to/tspannhw/google-coral-tpu-with-edge-devices-and-minifi-e6n)
- [https://github.com/tspannhw/minifi-grove-sensors](https://github.com/tspannhw/minifi-grove-sensors)
- [https://coral.ai/docs/enviro-board/get-started/](https://coral.ai/docs/enviro-board/get-started/)