Building a Connected Vehicle Data Management System using EMQX and Kafka
A few days back, I received an email from EMQX, a company specializing in facilitating data transfer and transformation through EMQX brokers and various other components. Their focus extends beyond edge computing to encompass cloud-based approaches as well. Given our familiarity with initiatives from companies like HiveMQ, which have been featured in numerous tutorials showcasing IoT use cases, it’s reasonable to assume that EMQX can support similar endeavors. To explore the intricacies of EMQX’s capabilities, this tutorial will dissect the article titled “Connected Cars and Automotive Connectivity,” authored by EMQX itself. Within this article, EMQX delves into a myriad of use cases, applications, challenges, and the methodologies employed to construct such platforms. Moreover, they provide a code example leveraging EMQX technologies alongside other renowned software, illustrating the creation of a straightforward streaming data platform tailored for connected vehicles. To better grasp the contents of this article, we’ve divided our analysis into three distinct sections below.
- Deciphering the EMQX Connected Vehicle System Architecture
- Expanding and Implementing a Straightforward Demo Environment
- Showcasing Demo Outcomes
1. Deciphering the EMQX Connected Vehicle System Architecture
EMQX provides a comprehensive cloud platform solution tailored for connected vehicles, as depicted below. In this setup, vehicles establish connections through MQTT to the EMQX cluster located in the cloud. Here, the data undergoes storage and management via subcomponents before being relayed through Kafka brokers to other software or platforms for visualization or deeper analysis.
In a production environment, the depicted system architecture can be implemented to gauge its scalability, flexibility, availability, and other operational aspects during execution. However, assembling all these components together in the cloud poses practical challenges. The example provided by EMQX within the article is relatively small in scope, focusing on a simple use case as shown below. Despite its apparent simplicity, the example encompasses core elements necessary for understanding the intricate system architecture.
As per usual, on the left side, MQTTX, a software akin to TeslaMate project, generates Tesla data. This data is then transmitted to the EMQX Enterprise software illustrated below, where MQTT brokers, rule engine, and data bridge are stationed. Additionally, emqx-exporter furnishes supplementary metrics via REST API to Prometheus, with Grafana visualizing these incoming metrics. On the flip side, a MQTT subscriber and Kafka broker are linked across various topics, particularly mqtt/simulate/# and my-vehicles, to ingest data from the EMQX Enterprise software.
Another view of the EMQX Enterprise and its connection with other components is visualized to exactly understand the role of EMQX.
2. Expanding and Implementing a Straightforward Demo Environment
As previously mentioned, the EMQX tutorial furnishes an illustrative example through the architecture diagram depicted as “Tutorial System Architecture” above. This example aims to showcase the practical implementation of a connected vehicle use case leveraging EMQX tools in conjunction with other industry-standard software tools. Within this system architecture, EMQX plays a pivotal role in several key aspects: it facilitates data collection, distributes data to nodes within the node cluster, applies rules based on incoming data across MQTT topics, transfers processed data to the Kafka Cluster via relevant Kafka connectors, and offers data retrieval through a REST API for consumption of stored data. The depicted REST operations on the right side underscore potential scenarios for data consumption via Kafka clusters.
The example source code is accessible via this GitHub link. However, in our example, we’ve implemented some modifications. Firstly, we’ve opted to replace the Kafka image version from wurstmeister/kafka:2.12–2.3.0 to confluencetinc/cp-kafka:5.5.3. This alteration was necessitated by compatibility issues between kafka-consumer and the Kafka version. Secondly, we’ve introduced a new kafka-consumer as a Python application to demonstrate how data can be retrieved from the Kafka cluster for subsequent usage. Beyond these changes, the docker-compose.yml file remains largely unchanged.
version: '3'
volumes:
grafana:
services:
emqx-enterprise:
image: emqx/emqx-enterprise:5.0.4
container_name: emqx
healthcheck:
test: ["CMD", "emqx", "ping"]
interval: 10s
timeout: 10s
retries: 12
ports:
- 1883:1883
- 18083:18083
volumes:
- ./emqx/cluster.hocon:/opt/emqx/data/configs/cluster.hocon
- ./emqx/api_secret:/opt/emqx/data/api_secret
environment:
EMQX_DASHBOARD__BOOTSTRAP_USERS_FILE: '"/opt/emqx/data/api_secret"'
depends_on:
- kafka
networks:
- emqx-net
emqx-exporter:
image: emqx/emqx-exporter
container_name: emqx-exporter
ports:
- 8085:8085
command:
- '--emqx.nodes=emqx-enterprise:18083'
- '--emqx.auth-username=76668f8a2003d597'
- '--emqx.auth-password=CRCDB6lxxzN58e5HoD82llBC0Erg1TVZIAUsdTjPU7N'
depends_on:
emqx-enterprise:
condition: service_healthy
networks:
- emqx-net
mqttx-simulate:
image: emqx/mqttx-cli:v1.9.3
container_name: mqttx-simulate
command:
- /bin/sh
- -c
- |
mqttx simulate -sc tesla -c 20 -h emqx-enterprise
depends_on:
emqx-enterprise:
condition: service_healthy
networks:
- emqx-net
mqttx:
image: emqx/mqttx-cli:v1.9.3
container_name: mqttx
command:
- /bin/sh
- -c
- |
mqttx sub -t 'mqttx/simulate/#' -v -h emqx-enterprise
depends_on:
emqx-enterprise:
condition: service_healthy
networks:
- emqx-net
prometheus:
image: prom/prometheus
container_name: prometheus
ports:
- 9090:9090
volumes:
- ./prometheus/prometheus.yaml:/etc/prometheus/prometheus.yml
networks:
- emqx-net
grafana:
image: grafana/grafana:9.3.2
container_name: grafana
ports:
- 3000:3000
environment:
- GF_SECURITY_ADMIN_PASSWORD=public
volumes:
- grafana:/grafana-dashboard-emqx5-ee
- ./grafana-provisioning/dashboard.yaml:/etc/grafana/provisioning/dashboards/dashboard.yaml
- ./grafana-provisioning/datasource.yaml:/etc/grafana/provisioning/datasources/datasource.yaml
networks:
- emqx-net
kafka:
image: confluentinc/cp-kafka:5.5.3
container_name: kafka
ports:
- "9092:9092"
- "29092:29092"
environment:
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,PLAINTEXT_HOST://0.0.0.0:29092
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_CREATE_TOPICS: "my-vehicles:1:1"
depends_on:
- zookeeper
networks:
- emqx-net
zookeeper:
image: wurstmeister/zookeeper
container_name: zookeeper
ports:
- 2181:2181
environment:
ZOOKEEPER_CLIENT_PORT: 2181
networks:
- emqx-net
kafka-consumer:
build:
context: ./consumer_app
dockerfile: Dockerfile
container_name: kafka-consumer
restart: on-failure
depends_on:
- kafka
environment:
- KAFKA_BOOTSTRAP_SERVERS=kafka:9092
- KAFKA_TOPIC=my-vehicles
networks:
- emqx-net
networks:
emqx-net:
name: emqx-netIf you have a program such as Docker or Rancher capable of executing containers, you can initiate the containers by running docker-compose up. Please be patient during the startup process, as it may require a significant amount of time to initialize all components.
3. Showcasing Demo Outcomes
Assuming all components are running smoothly, we should begin receiving or observing incoming vehicle telemetry within the data consumers. Additionally, we should be able to monitor the metrics of the EMQX broker through Grafana software via Prometheus.
Traditional approach using kafka-consumer.sh
Get ca’s stories in your inbox
Join Medium for free to get updates from this writer.
In the original tutorial provided by EMQX, the kafka-consumer.sh file is included to showcase the data transmitted from the Kafka broker. An alternative method is to directly execute the kafka-console-consumer command within the Kafka Docker container, as demonstrated below in the terminal.
docker exec -it kafka \
/bin/bash -c "kafka-console-consumer --topic my-vehicles --bootstrap-server localhost:9092 --from-beginning"You will receive dozens of messages, and one of it is seen below:
{"car_id":"NY1VL087USDN12933","display_name":"Sherwood's Tesla","model":"X","trim_badging":"maxime","exterior_color":"lime","wheel_type":"vel","spoiler_type":"placeat","geofence":"Macon-Bibb County","state":"online","since":"2024-04-08T12:06:55.205Z","healthy":true,"version":"7.2.1","update_available":true,"update_version":"6.5.2","latitude":"-53.5067","longitude":"-22.9552","shift_state":"D","power":-6410,"speed":135,"heading":9,"elevation":1103,"locked":false,"sentry_mode":false,"windows_open":true,"doors_open":false,"trunk_open":false,"frunk_open":true,"is_user_present":false,"is_climate_on":false,"inside_temp":44.4,"outside_temp":9.7,"is_preconditioning":false,"odometer":695109,"est_battery_range_km":87.3,"rated_battery_range_km":221.1,"ideal_battery_range_km":306,"battery_level":98,"usable_battery_level":84,"plugged_in":false,"charge_energy_added":19.87,"charge_limit_soc":37,"charge_port_door_open":false,"charger_actual_current":64.75,"charger_power":79,"charger_voltage":238,"charge_current_request":34,"charge_current_request_max":34,"scheduled_charging_start_time":"2025-09-12T13:46:36.522Z","time_to_full_charge":6.3,"tpms_pressure_fl":2.2,"tpms_pressure_fr":2.9,"tpms_pressure_rl":3.3,"tpms_pressure_rr":3.3,"timestamp":1712592280961}As mentioned earlier, the dataset is generated based on the TeslaMate project, and the data model’s JavaScript implementation can be accessed via the provided link. The model is built using the faker JS library, a widely recognized tool. However, delving into the specifics of this library is beyond the scope of our current discussion.
Retrieving message via MQTT Consumer
The second data consumer is the MQTT subscriber, which retrieves the generated data from the topic mqttx/simulate/#. All received messages are then printed out on the terminal, as illustrated below.
Another approach is to directly utilize the MQTTX application and subscribe to the same topic. This allows all telemetry messages to be conveniently displayed on the user interface of the MQTTX app.
Observing messages in python consumer-app
The third consumer of telemetry data is our Python consumer application. Although its function aligns with that of the first consumer, it provides increased programming flexibility, facilitating connections to supplementary services. While Kafka offers robust plugins for data storage and forwarding, our Python consumer application offers enhanced adaptability. Below are the console outputs of the consumer application.
Displaying EMQX Metrics in Grafana Dashboard
In addition to telemetry data, monitoring the status of the EMQX Enterprise application is crucial. This enables us to track resource consumption, including CPU, memory, incoming and outgoing messages, and more. Grafana allows us to visualize this aggregated data. By utilizing the available dashboard in Grafana labs, we can seamlessly import the JSON model directly into the Grafana tool. To realize it, we first access the Grafana web interface via http://localhost:3000/ URL address, and enter the required credentials (u: admin, pw: public) , and then navigate to http://localhost:3000/dashboards where you should click Go to Folder .
Since we will import the JSON model, we click Import and then
upload JSON file downloaded from Grafana labs.
The last step is to select Prometheus as EMQX Prometheus,
and voilà, here the results once the import is successfully done. We can see the active connections, topics, subscriptions, total received/sent messages, and many other useful metrics.
Source Code Link: https://github.com/cemakpolat/emqx-mqtt-to-kafka
Summary
In this comprehensive tutorial, we delved into an EMQX tutorial covering various components such as EMQX Enterprise, MQTT publisher and subscriber, Kafka broker, Kafka-consumer, Prometheus, and Grafana software. Beyond just enhancing the demonstration, we provided an in-depth exploration of the system architecture of EMQX Enterprise software, drawing comparisons with the demo setup for clarity. To conclude, we meticulously executed the demo setup and disseminated the outcomes. It is my sincere hope that this tutorial proves valuable in aiding someone’s endeavors.
