FLaNK: Using Apache Kudu As a Cache For FDA Updates
Tags: apachenifi
Published: 2021-01-06 20:55:35

---
title: FLaNK:  Using Apache Kudu As a Cache For FDA Updates
published: true
date: 2021-01-06 20:54:00 UTC
tags: apachenifi
canonical_url: 
---

##  FLaNK:  Using Apache Kudu As a Cache For FDA Updates

[![](https://1.bp.blogspot.com/-2GxTVOW80T4/X_YdKaJpAYI/AAAAAAAAcO0/_pI4S4zhU14Y3U1uJBOXDmHsVZTWPp2fwCLcBGAsYHQ/w642-h640/dailymedflow0.png)](https://1.bp.blogspot.com/-2GxTVOW80T4/X_YdKaJpAYI/AAAAAAAAcO0/_pI4S4zhU14Y3U1uJBOXDmHsVZTWPp2fwCLcBGAsYHQ/s1486/dailymedflow0.png)

[![](https://1.bp.blogspot.com/-Tg9_kh96xcQ/X_YdP-rV02I/AAAAAAAAcPo/LgI7QpLUrnkWS-Ikn5KPzMZKNXNd7soBQCLcBGAsYHQ/w640-h208/updateAttribute.png)](https://1.bp.blogspot.com/-Tg9_kh96xcQ/X_YdP-rV02I/AAAAAAAAcPo/LgI7QpLUrnkWS-Ikn5KPzMZKNXNd7soBQCLcBGAsYHQ/s1263/updateAttribute.png)

1. **InvokeHTTP** :   We invoke the RSS feed to get our data.
2. **QueryRecord** :   Convert RSS to JSON
3. **SplitJson** :  Split one file into individual records.   (Should refactor to **ForkRecord** )
4. **EvaluateJSONPath** :  Extract attributes you need.
5. **ProcessGroup** for SPL Processing

[![](https://1.bp.blogspot.com/-F3nzF98YPiM/X_YdKa8CUyI/AAAAAAAAcO8/UTDsoiUO28cZBW_oLFRJlCeIywzMImL6wCLcBGAsYHQ/w640-h328/dailymedflow1.png)](https://1.bp.blogspot.com/-F3nzF98YPiM/X_YdKa8CUyI/AAAAAAAAcO8/UTDsoiUO28cZBW_oLFRJlCeIywzMImL6wCLcBGAsYHQ/s2048/dailymedflow1.png)

  

  

We call and check the RSS feed frequently and we parse out the records from the RSS(XML) feed and check against our cache.   We use Cloudera's Real-Time Datahub with Apache Impala/Apache Kudu as our cache to see if we already received a record for that.   If it hasn't arrived yet, it will be added to the Kudu cache and processed.
  

**SPL Processing**

  

We use the extracted SPL to grab a detailed record for that SPL using **InvokeHTTP**.

  

[![](https://1.bp.blogspot.com/-7h-BW-pdN20/X_YdK6cuuxI/AAAAAAAAcPA/ivl58GGC_TUVpOC0TpKXFwp9udQaLTS7gCLcBGAsYHQ/w640-h226/invokehttp.png)](https://1.bp.blogspot.com/-7h-BW-pdN20/X_YdK6cuuxI/AAAAAAAAcPA/ivl58GGC_TUVpOC0TpKXFwp9udQaLTS7gCLcBGAsYHQ/s1225/invokehttp.png)

  

  

We have created an Impala/Kudu table to cache this information with Apache Hue.
  

[![](https://1.bp.blogspot.com/-obVWUq9SYRI/X_YdKflcPSI/AAAAAAAAcO4/ECkh-aNiTbgU-KnBXbrj41rAEP4i7zndgCLcBGAsYHQ/w640-h324/huetablebrowser.png)](https://1.bp.blogspot.com/-obVWUq9SYRI/X_YdKflcPSI/AAAAAAAAcO4/ECkh-aNiTbgU-KnBXbrj41rAEP4i7zndgCLcBGAsYHQ/s2048/huetablebrowser.png)

  

[![](https://1.bp.blogspot.com/-Eo7N-wY2kOA/X_YdOw5rH1I/AAAAAAAAcPg/UrlchuWZQLY_C17rHCy8Q-ZSMDm_hp8CgCLcBGAsYHQ/w640-h320/splKuduCacheHue.png)](https://1.bp.blogspot.com/-Eo7N-wY2kOA/X_YdOw5rH1I/AAAAAAAAcPg/UrlchuWZQLY_C17rHCy8Q-ZSMDm_hp8CgCLcBGAsYHQ/s2011/splKuduCacheHue.png)

  

[![](https://1.bp.blogspot.com/-hsmHNUuODI8/X_YdPSRzdyI/AAAAAAAAcPk/FglwzjSTPpYm1epyROuHHGNFk_0e4QGRwCLcBGAsYHQ/w640-h560/spltabledefinition.png)](https://1.bp.blogspot.com/-hsmHNUuODI8/X_YdPSRzdyI/AAAAAAAAcPk/FglwzjSTPpYm1epyROuHHGNFk_0e4QGRwCLcBGAsYHQ/s936/spltabledefinition.png)

  

We use the **LookupRecord** to read from our cache.

  

 ![](https://1.bp.blogspot.com/-Z0bNcQEFiXE/X_YdLYdSMGI/AAAAAAAAcPE/TFpclxTsYM0LQitNKzn1h0pwrn8nr5ABgCLcBGAsYHQ/w640-h278/lookuprecord.png)

  

  

If we don't have that value yet, we send it to the table for an UPSERT.

  

[![](https://1.bp.blogspot.com/-8nSqIir2Law/X_YdOY6yJmI/AAAAAAAAcPc/jeI7dFHXpnEded3lf6Qn6Ks31vsxh-s8gCLcBGAsYHQ/w640-h376/putkudu.png)](https://1.bp.blogspot.com/-8nSqIir2Law/X_YdOY6yJmI/AAAAAAAAcPc/jeI7dFHXpnEded3lf6Qn6Ks31vsxh-s8gCLcBGAsYHQ/s1034/putkudu.png)

  

We send our raw data as XML/RSS to HDFS for archival use and audits.

  

[![](https://1.bp.blogspot.com/-OIsHlERn9u4/X_YdN6x3kjI/AAAAAAAAcPY/lbREFyGO7EUU6-6M0dQK1ashNu53R9UmgCLcBGAsYHQ/w640-h294/puthdfs.png)](https://1.bp.blogspot.com/-OIsHlERn9u4/X_YdN6x3kjI/AAAAAAAAcPY/lbREFyGO7EUU6-6M0dQK1ashNu53R9UmgCLcBGAsYHQ/s666/puthdfs.png)

  

We can see the results of our flow in Apache Atlas to see full governance of our solution.

[![](https://1.bp.blogspot.com/-w_so3HbvepU/X_YdNdYvjHI/AAAAAAAAcPU/2hyup97ySTAjD7Bcr4DDdIxebNKlMIOFACLcBGAsYHQ/w640-h272/putKuduSPL.png)](https://1.bp.blogspot.com/-w_so3HbvepU/X_YdNdYvjHI/AAAAAAAAcPU/2hyup97ySTAjD7Bcr4DDdIxebNKlMIOFACLcBGAsYHQ/s1717/putKuduSPL.png)

So with some simple NiFi flow, we can ingest all the new updates to DailyMed and not reprocess anything that we already have.    

## Resources

- [https://dailymed.nlm.nih.gov/dailymed/](https://dailymed.nlm.nih.gov/dailymed/)
- [https://github.com/tspannhw/EverythingApacheNiFi](https://github.com/tspannhw/EverythingApacheNiFi)
- [https://dzone.com/articles/smart-stocks-with-nifi-kafka-flink-sql](https://dzone.com/articles/smart-stocks-with-nifi-kafka-flink-sql)
- [https://github.com/tspannhw/ApacheConAtHome2020/tree/main/flows/DailyMed](https://github.com/tspannhw/ApacheConAtHome2020/tree/main/flows/DailyMed)