Auto saved by Logseq
This commit is contained in:
@@ -0,0 +1,25 @@
|
||||
# 2021-01-22-1507-SE4AS-Suggerimenti-Fine-Corso
|
||||
|
||||
**Cristina**
|
||||
- Little demos to do together would be great to have.
|
||||
- Start projects earlier
|
||||
- to set the requirements together (for instance i December)
|
||||
- and later (with the requirements finalized in January) to develop the project individually
|
||||
|
||||
|
||||
**Giuseppe**
|
||||
- Lab session is missing.
|
||||
|
||||
|
||||
**Emanuele**
|
||||
- The example of the Linux distribution maybe should be presented again later in the course.
|
||||
|
||||
|
||||
**Alessandro**
|
||||
- Recording of the lessons was a great opportunity for the students.
|
||||
|
||||
|
||||
**Scholastique**
|
||||
- She liked the course.
|
||||
- Recording the lectures was great.
|
||||
- Improving practical part.
|
||||
@@ -0,0 +1,26 @@
|
||||
title:: Planning_SE4AS_20-21
|
||||
|
||||
# Remaining lectures SE4AS
|
||||
|
||||
[Calendario SE4AS aa 20/21 - Google Sheets](https://docs.google.com/spreadsheets/d/1wKZiTDArTpo9TYrgDDfP61Hk33CfB6bMnmA7eGb-XRY/edit#gid=358768655)
|
||||
- 8/1/2021 [DONE]
|
||||
- Eclipse Smart Home / OpenHab
|
||||
- 12/1/2021
|
||||
- Installation OpenHub on the student machines
|
||||
- Decision Function / Evaluation issues
|
||||
- 15/1/2021
|
||||
- Exemplars:
|
||||
- S. Y. Shin, S. Nejati, M. Sabetzadeh, L. C. Briand, C. Arora, e F. Zimmer, «Dynamic adaptation of software-defined networks for IoT systems: a search-based approach», in _Proceedings of the IEEE/ACM 15th International Symposium on Software Engineering for Adaptive and Self-Managing Systems_, Seoul Republic of Korea, giu. 2020, pagg. 137–148, doi: [10.1145/3387939.3391603](https://doi.org/10.1145/3387939.3391603).
|
||||
- D. Weyns, M. U. Iftikhar, D. Hughes, e N. Matthys, «Applying Architecture-Based Adaptation to Automate the Management of Internet-of-Things», in _Software Architecture_, Cham, 2018, pagg. 49–67, doi: [10.1007/978-3-030-00761-4\_4](https://doi.org/10.1007/978-3-030-00761-4_4).
|
||||
- M. Provoost e D. Weyns, «DingNet: A Self-Adaptive Internet-of-Things Exemplar», in _2019 IEEE/ACM 14th International Symposium on Software Engineering for Adaptive and Self-Managing Systems (SEAMS)_, Montreal, QC, Canada, mag. 2019, pagg. 195–201, doi: [10.1109/SEAMS.2019.00033](https://doi.org/10.1109/SEAMS.2019.00033).
|
||||
- 19/1/2021
|
||||
- Exemplars:
|
||||
- P. Hnetynka, T. Bures, I. Gerostathopoulos, e J. Pacovsky, «Using component ensembles for modeling autonomic component collaboration in smart farming», in _Proceedings of the IEEE/ACM 15th International Symposium on Software Engineering for Adaptive and Self-Managing Systems_, Seoul Republic of Korea, giu. 2020, pagg. 156–162, doi: [10.1145/3387939.3391599](https://doi.org/10.1145/3387939.3391599).
|
||||
- I. Gerostathopoulos e E. Pournaras, «TRAPPed in Traffic? A Self-Adaptive Framework for Decentralized Traffic Optimization», in _2019 IEEE/ACM 14th International Symposium on Software Engineering for Adaptive and Self-Managing Systems (SEAMS)_, Montreal, QC, Canada, mag. 2019, pagg. 32–38, doi: [10.1109/SEAMS.2019.00014](https://doi.org/10.1109/SEAMS.2019.00014).
|
||||
- M. A. Cusumano, «Self-driving vehicle technology: progress and promises», _Commun. ACM_, vol. 63, n. 10, pagg. 20–22, set. 2020, doi: [10.1145/3417074](https://doi.org/10.1145/3417074).
|
||||
- Rehearsal Alessandro Sallese
|
||||
- Exemplars of Autonomous systems
|
||||
- 22/1/2021
|
||||
- Discussione progetti
|
||||
- Presentazione Alessandro Sallese
|
||||
- Riempire questionario
|
||||
@@ -0,0 +1,14 @@
|
||||
-
|
||||
#dataprocessing #IOT
|
||||
|
||||
- # Processed papers to prepare the slides
|
||||
- [[Stream Processing with IoT Data]]
|
||||
- [[IoT Cloud and Machine Learning The Building Blocks]]
|
||||
- # Structure to be worked
|
||||
- Data Pipeline Workflow [Slides](file:///C:/Users/david/Zotero/storage/BES28LR5/4_Data_Pipelines_For_ML1_MESA.pptx)
|
||||
![[Pasted image 20210106091126.png]]
|
||||
- Data Generation
|
||||
- Data Transformation
|
||||
- Data Ingestion DB (This part is well done in the slides )
|
||||
- Data Processing (this will be covered by slides 10-30 of [this](file:///C:/Users/david/Zotero/storage/34V2VWW7/Data%20Streaming%20in%20IoT%20and%20Big%20Data%20Analytics.pdf))
|
||||
- Data analytics (from the slides of the Bologna IoT course [Bologna IoT course](file:///C:/Users/david/Zotero/storage/WUYV8GM5/Bononi%20e%20Felice%20-%20IoT%20Sensor%20Data%20Processing.pdf)
|
||||
@@ -0,0 +1,333 @@
|
||||
page-type:: [[TEACHING/SE4IOT]]
|
||||
|
||||
- # Slides
|
||||
- Da slide 15 di [Bononi e Felice - IoT Sensor Data Processing 1.pdf](file:///C:/Users/david/Zotero/storage/WUYV8GM5/Bononi%20e%20Felice%20-%20IoT%20Sensor%20Data%20Processing.pdf)
|
||||
- Slides [Bononi e Felice - IoT Sensor Data Management.pdf](file:///C:/Users/david/Zotero/storage/4Z3ESYTU/Bononi%20e%20Felice%20-%20IoT%20Sensor%20Data%20Management.pdf)
|
||||
- # Installation 1.x
|
||||
According to [Downloads (influxdata.com)](https://portal.influxdata.com/downloads/)
|
||||
|
||||
```
|
||||
sudo apt-get install influxdb
|
||||
sudo service influxdb start
|
||||
```
|
||||
- # Play with 1.x (on WSL)
|
||||
```
|
||||
influx // this will start the client
|
||||
show databases
|
||||
create database myfirstdatabase
|
||||
use myfirstdatabase
|
||||
show measurements
|
||||
```
|
||||
|
||||
We get empty measurements. They are typically collected by agents like Telegraf,etc.
|
||||
|
||||
|
||||
We can manually insert measurements as follows:
|
||||
|
||||
```
|
||||
insert cpu,host=node1 value=10
|
||||
```
|
||||
|
||||
Now we can show the recently added measurement with
|
||||
```
|
||||
show measurements
|
||||
```
|
||||
|
||||
We can get the content of the *cpu* measurement as follows
|
||||
|
||||
```
|
||||
select * from cpu
|
||||
```
|
||||
|
||||
We can drop measurements with the following command
|
||||
|
||||
```
|
||||
drop measurement cpu
|
||||
```
|
||||
|
||||
Let's insert some measurements:
|
||||
|
||||
```
|
||||
insert cpu,host=node1 value=10
|
||||
insert cpu,host=node2 value=15
|
||||
insert cpu,host=node3 value=22
|
||||
```
|
||||
|
||||
Now we can do again
|
||||
```
|
||||
select * from cpu
|
||||
```
|
||||
|
||||
We can see all the series as follows:
|
||||
```
|
||||
show series
|
||||
```
|
||||
|
||||
We can do some filtering as follows:
|
||||
|
||||
```
|
||||
select * from cpu where host='node2'
|
||||
```
|
||||
|
||||
We can also do filtering based on time, e.g. I want to retrieve the data related to the last 5 minutes
|
||||
|
||||
```
|
||||
select * from cpu where time >= now() - 5m
|
||||
```
|
||||
|
||||
```
|
||||
select * from cpu where time >= now() - 5m and host='node3'
|
||||
```
|
||||
|
||||
And what happens if I do as follows?
|
||||
|
||||
```
|
||||
select * from cpu where time >= now() - 1m
|
||||
```
|
||||
- # Installation 2.x
|
||||
- Remember to stop influxdb 1.x
|
||||
- ```sudo service influxdb start
|
||||
sudo service influxdb stop
|
||||
```
|
||||
- ```
|
||||
wget https://dl.influxdata.com/influxdb/releases/influxdb2\_2.0.3\_amd64.deb
|
||||
sudo dpkg -i influxdb2\_2.0.3\_amd64.deb
|
||||
influxdb
|
||||
```
|
||||
|
||||
The web dashboard is available at [localhost:8086](http://localhost:8086/)
|
||||
-
|
||||
- # Play with 2.x (in [docker]
|
||||
- First of all you have to install it according to [Install InfluxDB | InfluxDB OSS v2 Documentation (influxdata.com)](https://docs.influxdata.com/influxdb/v2/install/?t=Docker) and run it
|
||||
- ```
|
||||
docker run --name influxdb -p 8086:8086 influxdb:2.7.4
|
||||
```
|
||||
- 
|
||||
- 5LW7B-5ySei2kPoLtPIwBFS0KfK-uAcnxYE04BoJhalwdzQjvRz5zi7bELcTNw_WDkfCUkJiDd-wvIIDvnYHjQ==
|
||||
- # Set up a configuration profile [NOT NEEDED HERE]
|
||||
```
|
||||
influx config create -n default \
|
||||
-u http://localhost:8086 \
|
||||
-o univaq \
|
||||
-t 5LW7B-5ySei2kPoLtPIwBFS0KfK-uAcnxYE04BoJhalwdzQjvRz5zi7bELcTNw_WDkfCUkJiDd-wvIIDvnYHjQ== \
|
||||
-a
|
||||
```
|
||||
|
||||
The token given above needs to be retrieved from [Tokens | Load Data | -- | InfluxDB](http://localhost:8086/orgs/fc77876718e82aaf/load-data/tokens)
|
||||
|
||||
|
||||
|
||||
To list the created configuration (**from within the docker container**)
|
||||
```
|
||||
influx config list
|
||||
```
|
||||
The bucket given above is selected from [Buckets | Load Data | -- | InfluxDB](http://localhost:8086/orgs/fc77876718e82aaf/load-data/buckets)
|
||||
|
||||
**Insert measaurements via NodeRed from MQTT**
|
||||
Publish to `/nodered/powerconsumption`
|
||||
|
||||
```
|
||||
{
|
||||
"device":"device1",
|
||||
"host":"NodeRed",
|
||||
"power":"373i"
|
||||
}
|
||||
```
|
||||
- # Insert measurements via Telegraf from MQTT
|
||||
[Install Telegraf | Telegraf Documentation (influxdata.com)](https://docs.influxdata.com/telegraf/v1/install/#download)
|
||||
|
||||
```
|
||||
# influxdata-archive_compat.key GPG Fingerprint: 9D539D90D3328DC7D6C8D3B9D8FF8E1F7DF8B07E
|
||||
wget -q https://repos.influxdata.com/influxdata-archive_compat.key
|
||||
echo '393e8779c89ac8d958f81f942f9ad7fb82a25e133faddaf92e15b16e6ac9ce4c influxdata-archive_compat.key' | sha256sum -c && cat influxdata-archive_compat.key | gpg --dearmor | sudo tee /etc/apt/trusted.gpg.d/influxdata-archive_compat.gpg > /dev/null
|
||||
echo 'deb [signed-by=/etc/apt/trusted.gpg.d/influxdata-archive_compat.gpg] https://repos.influxdata.com/debian stable main' | sudo tee /etc/apt/sources.list.d/influxdata.list
|
||||
sudo apt-get update && sudo apt-get install telegraf
|
||||
```
|
||||
- 
|
||||
- Telegraf Token
|
||||
```
|
||||
export INFLUX_TOKEN=YQ9t9PhYSeA6Ug0i44BInA-53gYAvq6hwa4_fVPI7MrXhUOugIFMl_o0_ib5Ulhr80-nFVtOmvr-QZWZtxt--Q==
|
||||
```
|
||||
- Run the Telegraf agent
|
||||
```
|
||||
telegraf --config http://localhost:8086/api/v2/telegrafs/0e03dad63d8aa000
|
||||
```
|
||||
- Publish this to the topic `sensors/powerconsumption`
|
||||
|
||||
```
|
||||
powerconsumption,device=device2,host=MyHost power=375i
|
||||
```
|
||||
|
||||
|
||||
There will be a Telegraf agent that will take such data and sends it to InfluxDB. To this end, Telegraf should be started (even on WSL) by using the following configuration file *telegraf_mqtt_to_influxdb2*
|
||||
```
|
||||
[[outputs.influxdb_v2]]
|
||||
## The URLs of the InfluxDB cluster nodes.
|
||||
##
|
||||
## Multiple URLs can be specified for a single cluster, only ONE of the
|
||||
## urls will be written to each interval.
|
||||
## urls exp: http://127.0.0.1:8086
|
||||
urls = ["http://localhost:8086"]
|
||||
|
||||
## Token for authentication.
|
||||
token = "WdpITesY4yOqb0dBy7n2WQgWSHPRP66O9mAzlEn7vjYBaCrnn7-bJEiKU5M0XqnU4Amf6So1XunJKKfyt9KGAA=="
|
||||
|
||||
## Organization is the name of the organization you wish to write to; must exist.
|
||||
organization = "univaq"
|
||||
|
||||
## Destination bucket to write into.
|
||||
bucket = "se4iot"
|
||||
|
||||
[[inputs.mqtt_consumer]]
|
||||
## Broker URLs for the MQTT server or cluster. To connect to multiple
|
||||
## clusters or standalone servers, use a seperate plugin instance.
|
||||
## example: servers = ["tcp://localhost:1883"]
|
||||
## servers = ["ssl://localhost:1883"]
|
||||
## servers = ["ws://localhost:1883"]
|
||||
servers = ["tcp://127.0.0.1:1883"]
|
||||
|
||||
## Topics that will be subscribed to.
|
||||
topics = [
|
||||
"sensors/#",
|
||||
]
|
||||
|
||||
## The message topic will be stored in a tag specified by this value. If set
|
||||
## to the empty string no topic tag will be created.
|
||||
# topic_tag = "topic"
|
||||
|
||||
## QoS policy for messages
|
||||
## 0 = at most once
|
||||
## 1 = at least once
|
||||
## 2 = exactly once
|
||||
##
|
||||
## When using a QoS of 1 or 2, you should enable persistent_session to allow
|
||||
## resuming unacknowledged messages.
|
||||
# qos = 0
|
||||
|
||||
## Connection timeout for initial connection in seconds
|
||||
# connection_timeout = "30s"
|
||||
|
||||
## Maximum messages to read from the broker that have not been written by an
|
||||
## output. For best throughput set based on the number of metrics within
|
||||
## each message and the size of the output's metric_batch_size.
|
||||
##
|
||||
## For example, if each message from the queue contains 10 metrics and the
|
||||
## output metric_batch_size is 1000, setting this to 100 will ensure that a
|
||||
## full batch is collected and the write is triggered immediately without
|
||||
## waiting until the next flush_interval.
|
||||
# max_undelivered_messages = 1000
|
||||
|
||||
## Persistent session disables clearing of the client session on connection.
|
||||
## In order for this option to work you must also set client_id to identify
|
||||
## the client. To receive messages that arrived while the client is offline,
|
||||
## also set the qos option to 1 or 2 and don't forget to also set the QoS when
|
||||
## publishing.
|
||||
# persistent_session = false
|
||||
|
||||
## If unset, a random client ID will be generated.
|
||||
# client_id = ""
|
||||
|
||||
## Username and password to connect MQTT server.
|
||||
# username = "telegraf"
|
||||
# password = "metricsmetricsmetricsmetrics"
|
||||
|
||||
## Optional TLS Config
|
||||
# tls_ca = "/etc/telegraf/ca.pem"
|
||||
# tls_cert = "/etc/telegraf/cert.pem"
|
||||
# tls_key = "/etc/telegraf/key.pem"
|
||||
## Use TLS but skip chain & host verification
|
||||
# insecure_skip_verify = false
|
||||
|
||||
## Data format to consume.
|
||||
## Each data format has its own unique set of configuration options, read
|
||||
## more about them here:
|
||||
## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
|
||||
data_format = "influx"
|
||||
```
|
||||
|
||||
Similar things can be done via command line as follows
|
||||
```sh
|
||||
export INFLUX_TOKEN=Ga8nVsXP4FAe5_M1a5j7uCa4zO_u_M9oUsO8wUSWh_wPbR3clc9ZTv420Li9adOVCPl1tGvn6hLHfI5gP7Lm5A==
|
||||
influx write -b se4iot -o univaq -p s 'powerconsumption,device=device2,host=CommandLine power=675i'
|
||||
```
|
||||
- sensors/powerconsumption
|
||||
- # Insert measurements via Java
|
||||
|
||||
```java
|
||||
package it.univaq.disim.se4iot.influxdbexample;
|
||||
|
||||
import java.time.Instant;
|
||||
|
||||
import com.influxdb.annotations.Column;
|
||||
import com.influxdb.annotations.Measurement;
|
||||
import com.influxdb.client.InfluxDBClient;
|
||||
import com.influxdb.client.InfluxDBClientFactory;
|
||||
import com.influxdb.client.WriteApi;
|
||||
import com.influxdb.client.domain.WritePrecision;
|
||||
import com.influxdb.client.write.Point;
|
||||
import com.influxdb.query.FluxTable;
|
||||
|
||||
public class InfluxDB2Example {
|
||||
|
||||
|
||||
public static void main(final String[] args) {
|
||||
|
||||
// You can generate a Token from the "Tokens Tab" in the UI
|
||||
String token = "5LW7B-5ySei2kPoLtPIwBFS0KfK-uAcnxYE04BoJhalwdzQjvRz5zi7bELcTNw_WDkfCUkJiDd-wvIIDvnYHjQ==";
|
||||
String bucket = "se4iot";
|
||||
String org = "univaq";
|
||||
|
||||
InfluxDBClient client = InfluxDBClientFactory.create("http://localhost:8086", token.toCharArray());
|
||||
|
||||
Point point = Point
|
||||
.measurement("powerconsumption")
|
||||
.addTag("device", "device3")
|
||||
.addField("power", 300)
|
||||
.time(Instant.now(), WritePrecision.NS);
|
||||
|
||||
try (WriteApi writeApi = client.getWriteApi()) {
|
||||
writeApi.writePoint(bucket, org, point);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
|
||||
In a dedicated Maven project with the following pom.xml
|
||||
|
||||
```xml
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<groupId>it.univaq.disim.se4iot.influxdbexample</groupId>
|
||||
<artifactId>it.univaq.disim.se4iot.influxdbexample</artifactId>
|
||||
<version>0.0.1-SNAPSHOT</version>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>com.influxdb</groupId>
|
||||
<artifactId>influxdb-client-java</artifactId>
|
||||
<version>1.8.0</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
<build>
|
||||
<sourceDirectory>src</sourceDirectory>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
<version>3.8.0</version>
|
||||
<configuration>
|
||||
<release>12</release>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
|
||||
</build>
|
||||
</project>
|
||||
```
|
||||
- # Some tutorials
|
||||
[(14) InfluxDB Tutorial - Complete Guide to getting started with InfluxDB - YouTube](https://www.youtube.com/watch?v=Vq4cDIdz_M8&ab_channel=DevOpsJourney)
|
||||
[(14) Introduction to InfluxDB 2 & Flux | Rawkode Live - YouTube](https://www.youtube.com/watch?v=LfWVZHuCCnE&ab_channel=DavidMcKay)
|
||||
[(14) Node-Red: inseriamo dati su InfluxDB - YouTube](https://www.youtube.com/watch?v=4Xeb3cOwrbs&ab_channel=L%27angolodelBIT)
|
||||
-
|
||||
- To remove all the data from the simple bucket
|
||||
- influx delete --bucket se4iot --org univaq --start 1970-01-01T00:00:00Z --stop 2100-01-01T00:00:00Z
|
||||
Binary file not shown.
@@ -0,0 +1,29 @@
|
||||
#Highlights
|
||||
|
||||
---
|
||||
|
||||
# **IoT Cloud and Machine Learning The Building Blocks**
|
||||
Date: 2021-01-05
|
||||
Time: 22:03
|
||||
URL: (https://www.opensourceforu.com/2020/10/iot-cloud-and-machine-learning-the-building-blocks/) [[Obsidian-Highlights]]
|
||||
Tags: #machinelearning #IOT #dataprocessing
|
||||
|
||||
---
|
||||
# Main building blocks of an ML infrastructure
|
||||
|
||||
Given below are the tools and frameworks needed for an Apache Spark based ML infrastructure.
|
||||
- _Spark 2:_ Comes up with all the necessary tools from the Spark ecosystem — Hadoop, Mlib, etc. You can get more details from https://spark.apache.org/docs/latest/.
|
||||
- _Hadoop:_ You can either install Spark2 with Hadoop or Hadoop as standalone.
|
||||
- _Python3/Scala/Java:_ This depends on what language you prefer to write ML programs
|
||||
- PostgreSQL/MongoDB: Install this if you have to store data into traditional databases other than Hadoop HDFS for future use and reference.
|
||||
- _MLib/Tensorflow/Keras/Scikit learn_: Choose from these ML libraries based on your needs.
|
||||
- _Data analytics tools:_ Choose these based on your needs.
|
||||
# Common ML use cases in IoT
|
||||
|
||||
Given below are a few common use cases based on the data received from devices (strictly based on my experience and may differ in your business case).
|
||||
- _Analysis of data patterns for a specific period_: As an example, if data comes from a temperature sensor, then the pattern of temperature data for a location where the device is installed for a day can be analysed for that specific period.
|
||||
- _Data missing/changes in duration/changes in pattern, etc:_ It is important to understand the missing data or changes in frequency because immediate action is required, or it can lead to potential errors in our analytics/forecasts.
|
||||
- _Inactivities or other ambiguities in data flow:_ These errors in data processing must be avoided.
|
||||
- _Difference between the forecasted and real data:_ This may lead to a correction in data models and trainings.
|
||||
- _User and location behaviour from device to device:_ Data for each device may differ if user and location behaviour contribute some points to the data.
|
||||
- _Frequency of maintenance and root causes for that:_ This may be specific to location, usage, transaction volume, etc.
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 279 KiB |
@@ -0,0 +1,16 @@
|
||||
title:: Planning_SE4IOT_20-21
|
||||
|
||||
-
|
||||
- # Remaining lectures SE4IOT
|
||||
- 8/1/2021 [OK]
|
||||
- Discussione progetti
|
||||
- Gateway (es. Kura) / Fatto vedere anche Kura installato mediante docker
|
||||
- [[IOT Data Processing and analytics]]
|
||||
- 13/1/2021 [OK]
|
||||
- [x] [[IOT Data Processing and analytics]]
|
||||
- 15/1/2021
|
||||
- [x] [[InfluxDB]]
|
||||
- 20/1/2021
|
||||
- Discussione progetti
|
||||
- Questionari
|
||||
-
|
||||
@@ -0,0 +1,79 @@
|
||||
#Highlights
|
||||
|
||||
---
|
||||
|
||||
# **Stream Processing with IoT Data**
|
||||
Date: 2021-01-05
|
||||
Time: 19:32
|
||||
URL: [Stream Processing with IoT Data: Best Practices & Techniques](https://www.confluent.io/blog/stream-processing-iot-data-best-practices-and-techniques/) [[Obsidian-Highlights]]
|
||||
Tags: #dataprocessing #IOT #streamprocessing
|
||||
|
||||
---
|
||||
# Intro
|
||||
|
||||
The rise of IoT devices means that we have to **collect**, **process**, and **analyze** orders of magnitude more data than ever before
|
||||
|
||||
...healthcare can now track patient health in real time and even provide on-demand care; manufacturing is able to understand the details of production lines and predict issues before they happen;
|
||||
|
||||
At Tesla, we unlock insights into our fleet by processing trillions of events per day from every part of the business, with just a handful of people.
|
||||
|
||||
This has helped Tesla become an industry leader by opening up new abilities, like being able to run complex predictions against our entire fleet to optimize efficiency or inform the next generation of manufacturing.
|
||||
|
||||
...is enabled by a system capable of ingesting, processing, storing, and serving these trillions of data points.
|
||||
|
||||
>IoT data streams look a lot like common web server log events. You have events being generated, sometimes at high volumes, and they need to be processed and either made available to downstream consumers or stored in databases.
|
||||
|
||||
...it turns out you have all of the usual “server log” challenges and then a slew of new ones to contend with as well. Instead of web servers that are probably within your network and under your control, you now have a large number of devices with variable connectivity, leading to bursty data and a long tail of firmware versions (you can’t just stop supporting some versions) with old data formats.
|
||||
|
||||
...some of these devices can go “insane” and start dumping piles of data on your infrastructure, which can feel like a Denial-of-Service (DoS) attack.
|
||||
|
||||
>Unfortunately, it gets even worse. Some of your data streams—especially if you have devices that are at all related to medical or health & safety—could be high priority and require very low latency.
|
||||
|
||||
These data streams can be mixed in with streams that are just high-volume “normal operations” streams used by analysts for evaluating and understanding the health of your device fleet. Now you have mixed service levels in a shared environment that you need to worry about.
|
||||
# Designing from first principles
|
||||
|
||||
A common organizational split is to have a firmware team that deals with making the device actually work and a server-side data team that handles the rest of the data pipeline, from collecting the events to processing, storing, and serving that data to the rest of the organization.
|
||||
|
||||
Before even deciding on technologies, the first question we should ask is: What kind of capabilities do I need from my system? There are a number of things you might want to include, but for IoT, here is a nice set of **core requirements**:
|
||||
- Durable storage
|
||||
- Easy horizontal scalability
|
||||
- High throughput
|
||||
- Low latency
|
||||
|
||||
...our challenge is to receive that data and quickly process it, making it available to downstream users so that they can produce the insights and improvements that really move a company forward.
|
||||
|
||||
The core piece of technology that enables us to meet all of these goals is Apache Kafka®.
|
||||
|
||||
Having settled on Kafka as the core of our infrastructure, we can start to sketch out some of the rest of the pieces of our processing system.
|
||||
|
||||
Let’s look at what we need to build around this core. We need to land the data from the devices into Apache Kafka, implement stream processing to make that raw data usable, and make the data accessible to others.
|
||||
## Raw data ingest
|
||||
|
||||
We need a small intermediary to abstract the Kafka client complexities from our devices, which often have computational and bandwidth constraints.
|
||||
|
||||
Great! Now that we have an API in place, we can do some simple routing and management of events.
|
||||
|
||||
...he API server also has one more job beyond routing: making sure that the data is durable before it confirms success to the client, so that the client knows if it should resend the data. However, for devices that have short-lived, on-device storage, or that are sending their “last gasp” data before they die, we might only have one chance to catch that data, so it is often vitally important that the data lands the first time, whenever possible
|
||||
## Handling large messages
|
||||
|
||||
One of the unique challenges with device data streams, as opposed to web server log data streams, is that it is normal for devices to become disconnected for a long while—easily months—and to then send a huge amount of data as they come back online.
|
||||
|
||||
I’ve seen Kafka happy with messages up to 1 MB, with very little tuning of configuration parameters needed. At the same time, I have heard of people regularly handling messages of 20 MB (e.g., [Vorstella](https://www.confluent.io/kafka-summit-san-francisco-2019/whats-inside-the-black-box-using-ml-to-tune-and-manage-kafka)) and upwards, but that requires significant tuning.
|
||||
|
||||
One approach is to split Kafka clusters by workload type: one for small messages, one for larger ones.
|
||||
|
||||
Here are two common, alternative approaches that we can take to handle these large messages within a single cluster:
|
||||
- Chunking up the message into pieces
|
||||
- Storing a message reference in Kafka to an external store
|
||||
|
||||
Using an external store, our API will write messages into Kafka with a small wrapper that includes the data (or the reference, for large messages), as well as some helpful details about the message:
|
||||
|
||||
```
|
||||
Message {
|
||||
string device_id;
|
||||
optional string reference;
|
||||
optional bytes body;
|
||||
}
|
||||
```
|
||||
|
||||
|
||||
Reference in New Issue
Block a user