FLaNK: Real-Time Transit Information For NY/NJ/CT (TRANSCOM)
Published: 2021-01-09 18:31:38

---
title: FLaNK:   Real-Time Transit Information For NY/NJ/CT (TRANSCOM)
published: true
date: 2021-01-09 18:31:00 UTC
tags: 
canonical_url: 
---

#  FLaNK:   Real-Time Transit Information For NY/NJ/CT (TRANSCOM)

**SOURCE** : XML/RSS REST ENDPOINT 

**FREQUENCY** :  Every Minute

**DESTINATIONS** :   HDFS, Kudu/Impala, Cloud, Kafka

  

[![](https://1.bp.blogspot.com/-IBiBeDez05E/X_n0FS-qicI/AAAAAAAAcSQ/iquXkIJIWwcPXFY4tLsOtORpb12b1xVYgCLcBGAsYHQ/w448-h496/nififlowoverview.png)](https://1.bp.blogspot.com/-IBiBeDez05E/X_n0FS-qicI/AAAAAAAAcSQ/iquXkIJIWwcPXFY4tLsOtORpb12b1xVYgCLcBGAsYHQ/s1038/nififlowoverview.png)
  

  

  

The main source of this real-time transit updates for New Jersey, New York and Connecticut is TRANSCOM.   I will read from this datasource every minute to know about real-time traffic events that occurring on the roads and transportation systems near me.   We will be reading the feed that is in XML/RSS format and parse out the hundreds of events that come with each minutes update.   

  

I want to store the raw XML/RSS file in S3/ADLS2/HDFS or GCS, that's an easy step.   I will also parse and enhance this data for easier querying and tracking.

  

I will add to all events a unique ID and a timestamp as the data is streaming by.   I will store my data in Impala/Kudu for fast queries and upserts.   I can then build some graphs, charts and tables with Apache Hue and Cloudera Visual Applications.   I will also publish my data as AVRO enhanced with a schema to Kafka so that I can use it from Spark, Kafka Connect, Kafka Streams and Flink SQL applications.

  

  

[![](https://1.bp.blogspot.com/-UEryo6V5aBQ/X_n0JdLIihI/AAAAAAAAcSY/fAEro2oxQ3sdJoPzTHPXgeZVJZb7rSP_ACLcBGAsYHQ/w640-h370/flow1.png)](https://1.bp.blogspot.com/-UEryo6V5aBQ/X_n0JdLIihI/AAAAAAAAcSY/fAEro2oxQ3sdJoPzTHPXgeZVJZb7rSP_ACLcBGAsYHQ/s2048/flow1.png)

  

1. **GenerateFlowFile** - optional scheduler
2. **InvokeHTTP** - call RSS endpoint
3. **PutHDFS** - store raw data to Object or File store on premise or in the cloud via HDFS / S3 / ADLSv2 / GCP / Ozone / ...
4.   **QueryRecord** - convert XML to JSON
5. **SplitJSON** - break out individual events

  

[![](https://1.bp.blogspot.com/-A5Otxw_i_So/X_n0JeXvd_I/AAAAAAAAcSU/t-_17LUmq-wmF4IGNSUnLmqRkV_dYaOHwCLcBGAsYHQ/w560-h819/flow2.png)](https://1.bp.blogspot.com/-A5Otxw_i_So/X_n0JeXvd_I/AAAAAAAAcSU/t-_17LUmq-wmF4IGNSUnLmqRkV_dYaOHwCLcBGAsYHQ/s1562/flow2.png)

  

1. **UpdateAttribute** - set schema name
2. **UpdateRecord** - generate an add a unique ID and timestamp
3. **UpdateRecord** - clean up the point field
4. **UpdateRecord** - remove garbage whitespace

  

[![](https://1.bp.blogspot.com/-q6fqFBPLVkc/X_n0JYzTRLI/AAAAAAAAcSc/CctjPWBS94Ydh9gBUaCOyn-jHN0gO1ksQCLcBGAsYHQ/w640-h294/flow3.png)](https://1.bp.blogspot.com/-q6fqFBPLVkc/X_n0JYzTRLI/AAAAAAAAcSc/CctjPWBS94Ydh9gBUaCOyn-jHN0gO1ksQCLcBGAsYHQ/s2617/flow3.png)
  

  

1. **PutKudu -** upsert new data to our Impala / Kudu table.
2. **RetryFlowFile** - retry if network or other connectivity issue.

  

**Send Messages to Kafka**

  

[![](https://1.bp.blogspot.com/-k3lDQP_1a8w/X_n1CtdSvSI/AAAAAAAAcSs/D9u2TtDBvngw0yyTVRQg-brZppCtbXOkQCLcBGAsYHQ/w640-h352/publishKafka1.png)](https://1.bp.blogspot.com/-k3lDQP_1a8w/X_n1CtdSvSI/AAAAAAAAcSs/D9u2TtDBvngw0yyTVRQg-brZppCtbXOkQCLcBGAsYHQ/s1084/publishKafka1.png)
  

[![](https://1.bp.blogspot.com/-hSnGerIHxu4/X_n1CmNrK6I/AAAAAAAAcSw/BjBpJnLBgnMV4jsDTNu7NP558XDibPW8ACLcBGAsYHQ/w640-h422/publishKafka2.png)](https://1.bp.blogspot.com/-hSnGerIHxu4/X_n1CmNrK6I/AAAAAAAAcSw/BjBpJnLBgnMV4jsDTNu7NP558XDibPW8ACLcBGAsYHQ/s943/publishKafka2.png)

  

  

Our flow has delivered many messages to our **transcomevents**  topic as schema attached Apache Avro formatted messages.  

  

  

[![](https://1.bp.blogspot.com/-3250H0wpIQI/X_n1Q2TKfHI/AAAAAAAAcS8/JgGVeh0mCFUAjZQAXdzBhRSrJi3WkTxLgCLcBGAsYHQ/w640-h312/smmdataavro.png)](https://1.bp.blogspot.com/-3250H0wpIQI/X_n1Q2TKfHI/AAAAAAAAcS8/JgGVeh0mCFUAjZQAXdzBhRSrJi3WkTxLgCLcBGAsYHQ/s2536/smmdataavro.png)
  

[![](https://1.bp.blogspot.com/-rsWHk6xIErs/X_n1Q6szG3I/AAAAAAAAcS0/oqxRXFAK7fkW3LIJPpuZbvLSb5vDOfwggCLcBGAsYHQ/w640-h116/smmdataoverview.png)](https://1.bp.blogspot.com/-rsWHk6xIErs/X_n1Q6szG3I/AAAAAAAAcS0/oqxRXFAK7fkW3LIJPpuZbvLSb5vDOfwggCLcBGAsYHQ/s2634/smmdataoverview.png)
  

[![](https://1.bp.blogspot.com/-KuHcz7dmJ9A/X_n1RUzJ43I/AAAAAAAAcTI/ek_ctYEradgWEr25DxyfIXqymAkuilADACLcBGAsYHQ/w640-h250/smmmetrics2.png)](https://1.bp.blogspot.com/-KuHcz7dmJ9A/X_n1RUzJ43I/AAAAAAAAcTI/ek_ctYEradgWEr25DxyfIXqymAkuilADACLcBGAsYHQ/s2837/smmmetrics2.png)
  

[![](https://1.bp.blogspot.com/-5fIqnj2LRfY/X_n1ROdduAI/AAAAAAAAcTA/KBmtyWA57lE5iTx9F719oweWPA-TAoS_QCLcBGAsYHQ/w640-h248/smmmetrics.png)](https://1.bp.blogspot.com/-5fIqnj2LRfY/X_n1ROdduAI/AAAAAAAAcTA/KBmtyWA57lE5iTx9F719oweWPA-TAoS_QCLcBGAsYHQ/s2844/smmmetrics.png)

  

SMM links into the Schema Registry and schema for this topic.
  

[![](https://1.bp.blogspot.com/-UWI2bleghjU/X_n1RRX6t7I/AAAAAAAAcTE/xPTcXVhJ-b8vlUEKWRbG9f5CsWvGzkJJQCLcBGAsYHQ/w640-h516/smmschema.png)](https://1.bp.blogspot.com/-UWI2bleghjU/X_n1RRX6t7I/AAAAAAAAcTE/xPTcXVhJ-b8vlUEKWRbG9f5CsWvGzkJJQCLcBGAsYHQ/s1683/smmschema.png)

  

We use a schema for validation and as a contract between consumers and producers of these traffic events.
  

[![](https://1.bp.blogspot.com/-TRhuEusA0tY/X_n1RuVeadI/AAAAAAAAcTM/Wg4YdKRuUOQy8MENOLDHEe5niDuHEClFgCLcBGAsYHQ/w640-h220/transcomschema.png)](https://1.bp.blogspot.com/-TRhuEusA0tY/X_n1RuVeadI/AAAAAAAAcTM/Wg4YdKRuUOQy8MENOLDHEe5niDuHEClFgCLcBGAsYHQ/s3028/transcomschema.png)

  

Since events are streaming into our Kafka topic and have a schema, we can query them with Continuous SQL with Flink SQL.  We can then run some Continuous ETL.

  

[![](https://1.bp.blogspot.com/-m3FSxVK5L9Q/X_n14g9ylvI/AAAAAAAAcTk/TiRxmiBJ1qEVfFPkEreM-uxFyf3i051bgCLcBGAsYHQ/w640-h312/flinksqltable.png)](https://1.bp.blogspot.com/-m3FSxVK5L9Q/X_n14g9ylvI/AAAAAAAAcTk/TiRxmiBJ1qEVfFPkEreM-uxFyf3i051bgCLcBGAsYHQ/s1304/flinksqltable.png)

  

[![](https://1.bp.blogspot.com/-JueBOznYiNc/X_n14hyc0qI/AAAAAAAAcTs/wi1ROej9uzUWm9qPI2gLI7PeeFFrddwqgCLcBGAsYHQ/w640-h342/flinksqlresults.png)](https://1.bp.blogspot.com/-JueBOznYiNc/X_n14hyc0qI/AAAAAAAAcTs/wi1ROej9uzUWm9qPI2gLI7PeeFFrddwqgCLcBGAsYHQ/s2048/flinksqlresults.png)

  

We could also consume this data with Structured Spark Streaming applications, Spring Boot apps, Kafka Streams, Stateless NiFi and Kafka Connect applications.  

  

We also stored our data in Impala / Kudu for permanent storage, ad-hoc queries, analytics, Cloudera Visualizations, reports, applications and more.

  

  

  

  

  

[![](https://1.bp.blogspot.com/-7bY9nt5XsRs/X_n14i6H_AI/AAAAAAAAcTo/KjquTStFxK8RTZDGF2FhRojoz9Rc9X0oACLcBGAsYHQ/w640-h546/hueevents.png)](https://1.bp.blogspot.com/-7bY9nt5XsRs/X_n14i6H_AI/AAAAAAAAcTo/KjquTStFxK8RTZDGF2FhRojoz9Rc9X0oACLcBGAsYHQ/s1994/hueevents.png)
  

[![](https://1.bp.blogspot.com/-BHOuagDGMO4/X_n1412AydI/AAAAAAAAcTw/Gtc0OKkG53srcxyCKMfh6E7GoSksFRv5wCLcBGAsYHQ/w640-h244/kudutable.png)](https://1.bp.blogspot.com/-BHOuagDGMO4/X_n1412AydI/AAAAAAAAcTw/Gtc0OKkG53srcxyCKMfh6E7GoSksFRv5wCLcBGAsYHQ/s1889/kudutable.png)
  

[![](https://1.bp.blogspot.com/-nFKuimlF-2g/X_n145kjMLI/AAAAAAAAcT0/FeKWRPHR6ygNRBPG8nyrZQZGPTjPbtbYQCLcBGAsYHQ/w640-h306/tablebrowser.png)](https://1.bp.blogspot.com/-nFKuimlF-2g/X_n145kjMLI/AAAAAAAAcT0/FeKWRPHR6ygNRBPG8nyrZQZGPTjPbtbYQCLcBGAsYHQ/s2567/tablebrowser.png)
  

[![](https://1.bp.blogspot.com/-98ZloJd381g/X_n140Fz07I/AAAAAAAAcT4/6p9cSrB6n4oZ1J0_7EWuWdIrQBbsRl6IgCLcBGAsYHQ/w640-h410/tabledetails.png)](https://1.bp.blogspot.com/-98ZloJd381g/X_n140Fz07I/AAAAAAAAcT4/6p9cSrB6n4oZ1J0_7EWuWdIrQBbsRl6IgCLcBGAsYHQ/s2048/tabledetails.png)
  

  

It is very easy to have fast data against our agile Cloud Data Lakehouse.

  

### Source Code

  

- [https://github.com/tspannhw/SmartTransit](https://github.com/tspannhw/SmartTransit)

  

### **Resources**

- [https://xcm.org/home](https://xcm.org/home)
- [https://data1.xcmdata.org/DEWeb/Pages/index.jsp](https://data1.xcmdata.org/DEWeb/Pages/index.jsp)
- [https://511nj.org/home](https://511nj.org/home)
- [https://github.com/tspannhw/ApacheConAtHome2020](https://github.com/tspannhw/ApacheConAtHome2020)
- [https://github.com/tspannhw/SmartTransit](https://github.com/tspannhw/SmartTransit)
- [https://github.com/tspannhw/SmartStocks](https://github.com/tspannhw/SmartStocks)