nifi-kafka-druid-water-level-data

You can execute a command on the Kafka broker to list the available topics as follows:

You can see that Kafka consists of one broker, and the topics stations and measurements have been created with eight partitions each.

To see some records sent to Kafka, run the following commands. You can change the number of records to print via the -c parameter.

Below is an example of the output of two records:

Below is an example of the output of three records:

The records of the two topics only contain the needed data. The measurement records contain a station_uuid for the measuring station. The relationship is illustrated below.

The reason for splitting the data up into two different topics is the improved performance. One more straightforward solution would be to use a single topic and produce records like the following:

Notice the two last attributes that differ from the previously shown stations records. The obvious downside is that every measurement (multiple millions of it) has to contain all the data known about the station it was measured at. This often leads to transmitting and storing duplicated information, e.g., the longitude of a station, resulting in increased network traffic and storage usage. The solution is only to send a station’s known/needed data or measurement data. This process is called data normalization. The downside is that when analyzing the data, you need to combine the records from multiple tables in Druid (stations and measurements).

If you are interested in how many records have been produced to the Kafka topic so far, use the following command. It will print the last record produced to the topic partition, formatted with the pattern specified in the -f parameter. The given pattern will print some metadata of the record.

If you calculate 1,324,098 records * 8 partitions, you end up with ~ 10,592,784 records. The output also shows that the last measurement record was produced at the timestamp 1680606104652, translating to Di 4. Apr 13:01:44 CEST 2023 (using the command date -d @1680606104).

You can look at the ingestion job running in NiFi by opening the endpoint https from your stackablectl stacklet list command output. You have to use the endpoint from your command output. In this case, it is https://172.18.0.2:30198. Open it with your favourite browser. Suppose you get a warning regarding the self-signed certificate generated by the Secret Operator (e.g. Warning: Potential Security Risk Ahead). In that case, you must tell your browser to trust the website and continue.

Log in with the username admin and password adminadmin.

As you can see, the NiFi workflow consists of lots of components. It is split into three main components:

You can zoom in by using your mouse and mouse wheel.

The left workflows works as follows:

The right side works similarly but is executed in an endless loop to stream the data in near-realtime. Double-click on the Get station list processor to show the processor details.

Head over to the tab PROPERTIES.

Here, you can see the setting HTTP URL, which specifies the download URL from where the JSON file containing the stations is retrieved. Close the processor details popup by clicking OK. You can also have a detailed view of the Produce station records processor by double-clicking it.

The Kafka connection details within this processor - like broker addresses and topic names - are specified. It uses the JsonTreeReader to parse the downloaded JSON and the JsonRecordSetWriter to split it into individual JSON records before writing it out. Double-click the Get 30 days historic data processor.

This processor fetched the historical data for every station. Click on the HTTP URL property.

The HTTP URL does contain the ${station_uuid} placeholder, which gets replaced for every station.

Double-click the PublishKafkaRecord_2_6 processor.

You can also see the number of produced records by right-clicking on PublishKafkaRecord_2_6 and selecting View status history.

After selecting Messages Send (5 mins) in the top right corner, you can see that ~10 million records were produced in ~5 minutes, corresponding to ~30k measurements/s. Keep in mind that the demo uses a single-node NiFi setup. The performance can be increased by using multiple nodes.

Regarding the NiFi resources, use the hamburger menu icon on the top right corner and select Node Status History.

The diagram shows the used heap size of the NiFi node. You can also select other metrics to show in the top right corner.

You can have a look at the ingestion jobs running in Druid by opening the Druid endpoint router-https from your stackablectl stacklet list command output (https://172.18.0.2:31616 in this case).

By clicking on Supervisors at the top you can see the running ingestion jobs.

After clicking on the magnification glass to the right side of the RUNNING supervisor, you can see additional information (here, the supervisor measurements was chosen). On the tab Statistics on the left, you can see the number of processed records as well as the number of errors.

The statistics show that Druid ingested 2435 records during the last minute and has already ingested ~30 million records in total. All records have been ingested successfully, indicated by having no processWithError, thrownAway or unparseable records.

The started ingestion jobs have automatically created the Druid data sources stations and measurements. You can see the available data sources by clicking on Datasources at the top.

The Avg. row size (bytes) column shows that a typical measurement record has 3 bytes, while a station record has 222, more than 50 times the size. So, by choosing two dedicated topics over a single topic, this demo saved 50x of storage and computation costs.

By clicking on segments under Availability for the measurements data source, you can see the segments of the data source. In this case, the measurements data source is partitioned by the measurement day, resulting in 33 segments.

Druid offers a web-based way of querying the data sources via SQL. To achieve this you first have to click on Query at the top.

You can now enter any arbitrary SQL statement, to e.g. list 10 stations run

To count the measurements per day run

The demo has created a Dashboard to visualize the water level data. To open it, click on the tab Dashboards at the top.

Click on the dashboard called Water level data. It might take some time until the dashboard renders all the included charts.

The dashboard Water level data consists of multiple charts. To list the charts, click on the tab Charts at the top.

Click on the Chart Measurements / hour. On the left side, you can modify the chart and click on Update Chart to see the effect.

You can see that starting from 2024/06/16 some stations didn’t measure or transmit their data. They started sending measurements again at 2024/06/17.

To look at the stations' geographical distribution, you have to click on the tab Charts at the top again. Afterwards, click on the chart Stations distribution.

The stations are, of course, placed alongside waterways. They are coloured by the waters they measure, so all stations alongside a body of water have the same colour. You can move and zoom the map with your mouse to interactively explore the map. You can, e.g. have a detailed look at the water Rhein.

Within Superset, you can create dashboards and run arbitrary SQL statements. On the top, click on the tab SQL → SQL Lab.

On the left, select the database druid, the schema druid and set See table schema to stations or measurements.

In the right textbox, enter the desired SQL statement. We need to join the two tables to get exciting results. Run the following query to determine the number of measurements the stations made:

You can also find the number of measurements for every body of water:

What might also be interesting is the average and current measurement of the stations:

Within Superset (or the Druid web interface), you can execute arbitrary SQL statements to explore the water level data.

You also can create additional charts and bundle them together in a Dashboard. Have a look at the Superset documentation on how to do that.

You can use the NiFi web interface to collect arbitrary data and write it to Kafka (it’s recommended to use new Kafka topics for that). Alternatively, you can use a Kafka client like kcat to create new topics and ingest data. Using the Druid web interface, you can start an ingestion job that consumes and stores the data in an internal data source. There is an excellent tutorial from Druid on how to do this. Afterwards, the data source can be analyzed within Druid and Superset, like the water level data.