[FLaNK] Smart Weather Websocket Application - Kafka Consumer
Published: 2020-11-10 21:15:24

---
title: [FLaNK] Smart Weather Websocket Application - Kafka Consumer
published: true
date: 2020-11-10 20:42:00 UTC
tags: 
canonical_url: 
---

#  [FLaNK] Smart Weather Websocket Application - Kafka Consumer

**Part 2 of 2**

[![](https://1.bp.blogspot.com/-l4T9LGZgSKI/X6rxAv1m2mI/AAAAAAAAb3g/9I9sSNLS7lEPm7H_eMgdFRoSTWZieH7KACLcBGAsYHQ/w640-h458/webui2.png)](https://1.bp.blogspot.com/-l4T9LGZgSKI/X6rxAv1m2mI/AAAAAAAAb3g/9I9sSNLS7lEPm7H_eMgdFRoSTWZieH7KACLcBGAsYHQ/s1659/webui2.png)
  

This is based on Koji Kawamura's excellent GIST: 

[https://gist.github.com/ijokarumawak/60b9ab2038ef906731ebf4c0eee97176](https://gist.github.com/ijokarumawak/60b9ab2038ef906731ebf4c0eee97176)

As part of my Smart Weather Application, I wanted to display weather information as it arrives in a webpage using web sockets.   Koji has an excellent NiFi flow that does it.   I tweaked it and add some things since I am not using Zeppelin.   I am hosting my webpage with NiFi as well.

[https://www.datainmotion.dev/2020/11/flank-smart-weather-applications-with.html](https://dev.to/tspannhw/flank-smart-weather-applications-with-flink-sql-2kfj)

We simply supply a webpage that makes a websocket connection to NiFi and NiFi keeps a cache in HBase to know what the client is doing.  This cache is updated by consuming from Kafka.   We can then feed events as they happen to the page.

[![](https://1.bp.blogspot.com/-rSSwMKV-Efo/X6rxILCNLOI/AAAAAAAAb3k/sMBJ_pWnhrs9ly7zDUBS32KCAokp3bo4gCLcBGAsYHQ/w640-h348/debugwebui.png)](https://1.bp.blogspot.com/-rSSwMKV-Efo/X6rxILCNLOI/AAAAAAAAb3k/sMBJ_pWnhrs9ly7zDUBS32KCAokp3bo4gCLcBGAsYHQ/s2048/debugwebui.png)
  

Here is the JavaScript for the web page interface to websockets:

| **<script>** |
| | function sendMessage(type, payload) { |
| | websocket.send(makeMessage(type, payload)); |
| | } |
| |   
 |
| | function makeMessage(type, payload) { |
| | return JSON.stringify({ |
| | 'type': type, |
| | 'payload': payload |
| | }); |
| | } |
| |   
 |
| | var wsUri = "ws://edge2ai-1.dim.local:9091/test"; |
| |   
 |
| | websocket = new WebSocket(wsUri); |
| | websocket.onopen = function(evt) { |
| |   
 |
| | sendMessage('publish', { |
| | "message": document.getElementById("kafkamessage") |
| | }); |
| |   
 |
| | }; |
| | websocket.onerror = function(evt) {console.log('ERR', evt)}; |
| | websocket.onmessage = function(evt) { |
| | var dataPoints = JSON.parse(evt.data); |
| |   
 |
| | var output = document.getElementById("results"); |
| | var dataBuffer = "<p>"; |
| | |
| | for(var i=0;i<dataPoints.length;i++) |
| | { |
| | dataBuffer += " <img src=\"" + dataPoints[i].icon\_url\_base + dataPoints[i].icon\_url\_name + "\">  " + dataPoints[i].location + |
| | dataPoints[i].station\_id + "@" + dataPoints[i].latitude + ":" + |
| | dataPoints[i].longitude + "@" + dataPoints[i].observation\_time + |
| | dataPoints[i].temperature\_string + "," + dataPoints[i].relative\_humidity + "," + |
| | dataPoints[i].wind\_string +"<br>"; |
| |   
 |
| | } |
| | output.innerHTML = output.innerHTML + dataBuffer + "</p><br>"; |
| | }; |
| |   
 |
| | **</script>**  
  
  
  
  
**Video Walkthrough:**    [https://www.twitch.tv/videos/797412192?es\_id=bbacb7cb39](https://www.twitch.tv/videos/797412192?es_id=bbacb7cb39)  
  
**Source Code:   ** [https://github.com/tspannhw/SmartWeather/tree/main](https://github.com/tspannhw/SmartWeather/tree/main)  
  
  
  
**Kafka Topic**  

[![](https://1.bp.blogspot.com/-0M6vsGi1aKU/X6rxeTWoC_I/AAAAAAAAb5I/pFhleli4cTslAfbae1olOE2emtxNg4kcwCLcBGAsYHQ/w640-h368/weathersmm.png)](https://1.bp.blogspot.com/-0M6vsGi1aKU/X6rxeTWoC_I/AAAAAAAAb5I/pFhleli4cTslAfbae1olOE2emtxNg4kcwCLcBGAsYHQ/s1838/weathersmm.png)
  
**weathernj Schema**  

[![](https://1.bp.blogspot.com/-H56xFtV4_tc/X6rxeISQNXI/AAAAAAAAb5E/qNUrdbt4cK05Rt7POZE82eKIufw-BNYPACLcBGAsYHQ/w640-h506/weatherschemas.png)](https://1.bp.blogspot.com/-H56xFtV4_tc/X6rxeISQNXI/AAAAAAAAb5E/qNUrdbt4cK05Rt7POZE82eKIufw-BNYPACLcBGAsYHQ/s1464/weatherschemas.png)
  
**The schema registry has a live Swagger interface to it's REST API**  
  
[![](https://1.bp.blogspot.com/-694MD8Wa5V8/X6rxcXAvRuI/AAAAAAAAb4o/ioRBfAFY8C8f2dRLcn2krVv8PX1cut8uACLcBGAsYHQ/w640-h442/swaggerschemareg.png)](https://1.bp.blogspot.com/-694MD8Wa5V8/X6rxcXAvRuI/AAAAAAAAb4o/ioRBfAFY8C8f2dRLcn2krVv8PX1cut8uACLcBGAsYHQ/s1756/swaggerschemareg.png)  
  
  
**NiFi Flow Overview**  

[![](https://1.bp.blogspot.com/-4UM942wil_I/X6rxc-DGz4I/AAAAAAAAb4w/nbAn1Xp-Y2wGW-QRrWRTSBvkh1FZq9UVQCLcBGAsYHQ/w640-h392/weatherProcessing.png)](https://1.bp.blogspot.com/-4UM942wil_I/X6rxc-DGz4I/AAAAAAAAb4w/nbAn1Xp-Y2wGW-QRrWRTSBvkh1FZq9UVQCLcBGAsYHQ/s1085/weatherProcessing.png)
  
  
**Ingest Via REST All US Weather Data from Zipped XML**  
  

[![](https://1.bp.blogspot.com/-Hh4edADA6yk/X6rxdkctG0I/AAAAAAAAb5A/XHQ1XWOtmz4jROXak6QRnDwt0DO_ECPUQCLcBGAsYHQ/w470-h640/weatheringest.png)](https://1.bp.blogspot.com/-Hh4edADA6yk/X6rxdkctG0I/AAAAAAAAb5A/XHQ1XWOtmz4jROXak6QRnDwt0DO_ECPUQCLcBGAsYHQ/s1028/weatheringest.png)
  

  

**As Data Streamings In, We Can Govern It**  
  
[![](https://1.bp.blogspot.com/-zdu90tpPoUA/X6rxaFsMXGI/AAAAAAAAb30/MQv-4xweVWAvkb9D-3gbWVx9107dfwXtQCLcBGAsYHQ/w640-h300/atlas.png)](https://1.bp.blogspot.com/-zdu90tpPoUA/X6rxaFsMXGI/AAAAAAAAb30/MQv-4xweVWAvkb9D-3gbWVx9107dfwXtQCLcBGAsYHQ/s1894/atlas.png)  
  

  

  

**Ingested Data is Validated Against It's Schema Then Pushed to Kafka as Avro**
  

[![](https://1.bp.blogspot.com/-mG84J1PlBxo/X6rxczqb07I/AAAAAAAAb4s/kkehreppNgkTFQOdnlHGLRtavl0brq43wCLcBGAsYHQ/w634-h640/validateAndProduce.png)](https://1.bp.blogspot.com/-mG84J1PlBxo/X6rxczqb07I/AAAAAAAAb4s/kkehreppNgkTFQOdnlHGLRtavl0brq43wCLcBGAsYHQ/s1035/validateAndProduce.png)
  
**We consume that Kafka data in store it in Kudu for analytics**  
  

[![](https://1.bp.blogspot.com/-beInpq3XvPk/X6rxbsWGRmI/AAAAAAAAb4Y/pJVZeaVjkhk277xd9oUK81VBznwqo88KQCLcBGAsYHQ/w640-h258/kafkatokudu.png)](https://1.bp.blogspot.com/-beInpq3XvPk/X6rxbsWGRmI/AAAAAAAAb4Y/pJVZeaVjkhk277xd9oUK81VBznwqo88KQCLcBGAsYHQ/s2464/kafkatokudu.png)
  

  

**We host a web page for our Websockets Application in NiFi with 4 simple processors.**  

  

  

[![](https://1.bp.blogspot.com/-G6SoGkDHtcY/X6rxbqHUVeI/AAAAAAAAb4c/gPG870nQMFQqTpD86X9iH4Wkzc55hPt2wCLcBGAsYHQ/w640-h522/nifiuihostingwebpage.png)](https://1.bp.blogspot.com/-G6SoGkDHtcY/X6rxbqHUVeI/AAAAAAAAb4c/gPG870nQMFQqTpD86X9iH4Wkzc55hPt2wCLcBGAsYHQ/s1411/nifiuihostingwebpage.png)
  
**Listen and Put Web Socket Messages Between NiFi Server and Web Application**  
  

[![](https://1.bp.blogspot.com/-knCOjrlzBvY/X6r5UdeS7kI/AAAAAAAAb5o/OxOpUiIyZq4L_A4P_kBD8N4o1voGQf_LACLcBGAsYHQ/w640-h448/listenToWebsockets.png)](https://1.bp.blogspot.com/-knCOjrlzBvY/X6r5UdeS7kI/AAAAAAAAb5o/OxOpUiIyZq4L_A4P_kBD8N4o1voGQf_LACLcBGAsYHQ/s1338/listenToWebsockets.png)
  

  

**Kafka Data is Cached for Websocket Applications**
  

[![](https://1.bp.blogspot.com/-EUIT9zF4lU8/X6rxbGZJz6I/AAAAAAAAb4Q/MUv9bQSaFA0B76hkvZra9T9S3RVwdSIhACLcBGAsYHQ/w640-h548/kafkatocache.png)](https://1.bp.blogspot.com/-EUIT9zF4lU8/X6rxbGZJz6I/AAAAAAAAb4Q/MUv9bQSaFA0B76hkvZra9T9S3RVwdSIhACLcBGAsYHQ/s1216/kafkatocache.png)
  
**Set the Port for WebSockets via Jetty Web Server**  
  
[![](https://1.bp.blogspot.com/-frSy5PM6kYk/X6rxbu3FkTI/AAAAAAAAb4U/YP6uuW0C_4ADgfeUxP8cm3BpnZHvh_INgCLcBGAsYHQ/w640-h84/listenwebsocket.png)](https://1.bp.blogspot.com/-frSy5PM6kYk/X6rxbu3FkTI/AAAAAAAAb4U/YP6uuW0C_4ADgfeUxP8cm3BpnZHvh_INgCLcBGAsYHQ/s830/listenwebsocket.png)  

[![](https://1.bp.blogspot.com/-VouKB9uJj9A/X6rxbItsAII/AAAAAAAAb4M/x1xzc1IWS8EVC39zYfnr-fMLlZoT09CQwCLcBGAsYHQ/w640-h558/jettywebsocket.png)](https://1.bp.blogspot.com/-VouKB9uJj9A/X6rxbItsAII/AAAAAAAAb4M/x1xzc1IWS8EVC39zYfnr-fMLlZoT09CQwCLcBGAsYHQ/s569/jettywebsocket.png)
**Use HBase As Our Cache**  

[![](https://1.bp.blogspot.com/-xjxGZ45cpJc/X6rxawwpJWI/AAAAAAAAb4I/jk0SxXyqe3ctej5F6Dtt-j6kE_m-1xp9wCLcBGAsYHQ/w640-h308/hbaseclient.png)](https://1.bp.blogspot.com/-xjxGZ45cpJc/X6rxawwpJWI/AAAAAAAAb4I/jk0SxXyqe3ctej5F6Dtt-j6kE_m-1xp9wCLcBGAsYHQ/s896/hbaseclient.png)
[![](https://1.bp.blogspot.com/-TuXZzUElZew/X6rxZw8yaGI/AAAAAAAAb3w/yWRTJmOpCugZ4OQu7cQy2fgx9McOrD5-QCLcBGAsYHQ/w640-h164/fetchCache.png)](https://1.bp.blogspot.com/-TuXZzUElZew/X6rxZw8yaGI/AAAAAAAAb3w/yWRTJmOpCugZ4OQu7cQy2fgx9McOrD5-QCLcBGAsYHQ/s873/fetchCache.png)  

[![](https://1.bp.blogspot.com/-aYoF9IPEQ5U/X6rxaodh9XI/AAAAAAAAb4A/cuzsucDWLOM9pBywVLYeKNUMWNTcZ_XtwCLcBGAsYHQ/w640-h220/hbaseCache.png)](https://1.bp.blogspot.com/-aYoF9IPEQ5U/X6rxaodh9XI/AAAAAAAAb4A/cuzsucDWLOM9pBywVLYeKNUMWNTcZ_XtwCLcBGAsYHQ/s856/hbaseCache.png)
  
  
  
**We can monitor our Flink SQL application from the Global Flink Dashboard**  
  

[![](https://1.bp.blogspot.com/-s5olCasYci8/X6rxasN5GUI/AAAAAAAAb38/z4FAByjT2LQWRovbVvoKFE4Hj-W5VqBpACLcBGAsYHQ/w640-h304/flinksql.png)](https://1.bp.blogspot.com/-s5olCasYci8/X6rxasN5GUI/AAAAAAAAb38/z4FAByjT2LQWRovbVvoKFE4Hj-W5VqBpACLcBGAsYHQ/s2202/flinksql.png)
  

  

[![](https://1.bp.blogspot.com/--qM02AP4h-c/X6rxajs1IVI/AAAAAAAAb4E/n4gGyt560E4ADsV_aKUoz6WN1hSpAFxBACLcBGAsYHQ/w640-h326/flinkglobal1.png)](https://1.bp.blogspot.com/--qM02AP4h-c/X6rxajs1IVI/AAAAAAAAb4E/n4gGyt560E4ADsV_aKUoz6WN1hSpAFxBACLcBGAsYHQ/s2048/flinkglobal1.png)  

  

  
**We can query our Weather data store in Apache Kudu via Apache Impala through Apache Hue **  

  

[![](https://1.bp.blogspot.com/-Rkq6E2nm5hQ/X6rxdpmDICI/AAAAAAAAb44/p8h9QaKAIig448v5ZKgmuEnVyPRR8lOxQCLcBGAsYHQ/w640-h470/weatherhue.png)](https://1.bp.blogspot.com/-Rkq6E2nm5hQ/X6rxdpmDICI/AAAAAAAAb44/p8h9QaKAIig448v5ZKgmuEnVyPRR8lOxQCLcBGAsYHQ/s1536/weatherhue.png)  

  

  

**Kudu Visualizations of Our Weather Data in Cloudera Visual Applications**

  

[![](https://1.bp.blogspot.com/-67KM4ZYWYuA/X6rxaK01UoI/AAAAAAAAb34/W9fZZhm5VckrGYk2b3LhjDESIxYae6dFgCLcBGAsYHQ/w640-h536/drilldownmap.png)](https://1.bp.blogspot.com/-67KM4ZYWYuA/X6rxaK01UoI/AAAAAAAAb34/W9fZZhm5VckrGYk2b3LhjDESIxYae6dFgCLcBGAsYHQ/s1402/drilldownmap.png)

[![](https://1.bp.blogspot.com/-_OLk7HJ_3kg/X6rxcTN3V_I/AAAAAAAAb4k/ndWjzuaEcKsrkjpTJQU-9TvpZ3NhzhzMQCLcBGAsYHQ/w640-h514/reports2.png)](https://1.bp.blogspot.com/-_OLk7HJ_3kg/X6rxcTN3V_I/AAAAAAAAb4k/ndWjzuaEcKsrkjpTJQU-9TvpZ3NhzhzMQCLcBGAsYHQ/s1420/reports2.png)
[![](https://1.bp.blogspot.com/-8QSt_zXVa9M/X6rxdqGiP4I/AAAAAAAAb48/NDDVDz25Eakp219mKW6dFgGx4pDA63dCgCLcBGAsYHQ/w640-h536/weatherapp2.png)](https://1.bp.blogspot.com/-8QSt_zXVa9M/X6rxdqGiP4I/AAAAAAAAb48/NDDVDz25Eakp219mKW6dFgGx4pDA63dCgCLcBGAsYHQ/s1407/weatherapp2.png) |