https://community.databricks.com/t5/technical-blog/real-time-vehicle-fleet-analytics-with-databricks-delta-live/ba-p/91422 <H1><SPAN>Introduction</SPAN></H1> <P><SPAN>Today’s modern vehicles generate large streams of data which can be harnessed in a data-driven manner to boost efficiency, optimize costs, and minimize our environmental impact in real-time. By leveraging this data across a fleet of vehicles at scale, we can achieve even greater benefits, enabling more effective fleet management and optimization.</SPAN></P> <P><SPAN>When it comes to building real-time streaming applications from IoT devices, including vehicle fleet telemetry and other edge devices</SPAN><SPAN>, </SPAN><A href="https://protobuf.dev/" target="_blank" rel="noopener"><SPAN>Protobuf (Protocol Buffers)</SPAN></A><SPAN> has emerged as a popular choice for companies. Its compact, efficient binary format ensures low latency and high throughput, making it ideal for handling critical data, enabling real-time decision-making, and optimizing operations.</SPAN></P> <P><SPAN>Traditional approaches to managing Protobuf events can often introduce a heavy burden on Data Engineering teams, requiring specialized skills that are rare and costly. This increases the ongoing expenses related to maintaining proto packages, system upgrades, and infrastructure. Additionally, data integrity issues during event replays add further complexity, making it challenging for businesses to manage streaming applications efficiently and cost-effectively. This highlights the need for a simple, reliable, and performant solution that minimizes the dependency on technical resources.&nbsp;</SPAN></P> <P><SPAN>This blog explores how you can harness the capabilities of </SPAN><A href="https://docs.databricks.com/en/delta-live-tables/serverless-dlt.html" target="_blank" rel="noopener"><SPAN>Serverless Delta Live Tables (DLT)</SPAN></A><SPAN> pipelines to transform raw IoT telemetry data from fleets into powerful insights. With </SPAN><A href="https://docs.databricks.com/en/structured-streaming/protocol-buffers.html" target="_blank" rel="noopener"><SPAN>native support</SPAN></A><SPAN>, Databricks greatly simplifies the process of serializing and de-serializing Protobuf events to enable these insights. Furthermore, Serverless ensures your data teams can focus on high-value activities like extracting insights while Databricks efficiently manages compute resources, including optimizing and scaling compute for workloads.&nbsp;</SPAN></P> <P>&nbsp;</P> <P><FONT size="5"><STRONG>Table of Contents</STRONG></FONT></P> <P><LI-TOC indent="15" liststyle="disc" maxheadinglevel="2"></LI-TOC></P> <H2>&nbsp;</H2> <H1><SPAN>Use Case: Optimize operations for your entire vehicle fleet&nbsp;</SPAN></H1> <P><SPAN>Company XYZ is a hypothetical logistics provider created to demonstrate real-time vehicle fleet analytics with Databricks Delta Live Tables. They operate thousands of vehicles across various routes daily, facing challenges in optimizing fuel efficiency and monitoring vehicle utilization. Excessive idle times and unmonitored fuel consumption can lead to higher operational costs.</SPAN></P> <P><SPAN>To address these challenges, they leverage real-time data analytics using <A href="https://www.databricks.com/product/delta-live-tables" target="_blank" rel="noopener">Databricks Delta Live Tables (DLT)</A> to gain actionable insights from their fleet telemetry data, enabling them to optimize vehicle usage and reduce fuel expenses effectively.</SPAN></P> <P><SPAN>IoT devices in vehicles can transmit telemetry data such as location, speed, fuel level, and engine status. These data points can power insights into fuel efficiency and vehicle utilization, which are essential metrics for optimizing fleet operations.&nbsp;</SPAN></P> <P><SPAN>The high-level architecture diagram below illustrates the end-to-end implementation of a streaming solution using Kafka and the Databricks Serverless DLT pipeline.&nbsp;</SPAN></P> <P>&nbsp;</P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="High-level Architecture.jpg" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/11415i9EEB3BC3960C04C9/image-size/large?v=v2&amp;px=999" role="button" title="High-level Architecture.jpg" alt="High-level Architecture.jpg" /></span></P> <P>&nbsp;</P> <P><SPAN>A data producer generates fleet telemetry events into a Kafka topic in the Protobuf format. In real-world architectures, this data producer is replaced by the vehicle fleet telemetry IoT devices, which send events to a centralized system such as gRPC Server. This system packages and compacts the individual events into batch messages before transmitting them to Kafka. The Serverless DLT pipeline continuously ingests those events from the Kafka topic, deserializes them using the </SPAN><SPAN>from_protobuf</SPAN> <SPAN>function, and populates the following tables in the medallion architecture.</SPAN></P> <UL> <LI><SPAN>The </SPAN><FONT face="courier new,courier"><SPAN>bronze_fleet_events_raw</SPAN></FONT><SPAN> table stores the raw events received from the Kafka topic.</SPAN></LI> <LI><SPAN>The </SPAN><FONT face="courier new,courier"><SPAN>bronze_fleet_events</SPAN></FONT><SPAN> table stores the deserialized Protobuf messages.</SPAN></LI> <LI><SPAN>The </SPAN><FONT face="courier new,courier"><SPAN>silver_fleet_events</SPAN></FONT><SPAN> table is an SCD type-1 that handles event replays from edge devices to ensure data integrity.</SPAN></LI> <LI><SPAN>The gold materialized views </SPAN><FONT face="courier new,courier"><SPAN>gold_fuel_efficiency</SPAN></FONT> <SPAN>and</SPAN> <FONT face="courier new,courier"><SPAN>gold_vehicle_utilization</SPAN></FONT><SPAN>, aggregations built on top of the Silver table, are used for downstream reporting and analytics.</SPAN></LI> </UL> <P><SPAN>These Gold materialized views are critical for downstream analytics. They provide insights into fuel efficiency and vehicle utilization, which are essential metrics for optimizing fleet operations.&nbsp;</SPAN></P> <P>&nbsp;</P> <TABLE border="1" width="100%"> <TBODY> <TR> <TD width="100%" style="background-color: lightyellow; width: 100%;"> <P><STRONG>Note!</STRONG></P> <P><SPAN>This blog uses Kafka to efficiently manage real-time fleet telemetry data, ensuring reliable streaming and handling of Protobuf-serialized messages.&nbsp;</SPAN></P> </TD> </TR> </TBODY> </TABLE> <P>&nbsp;</P> <H1><SPAN>Serverless Delta Live Tables</SPAN></H1> <P><SPAN>When you build your DLT pipeline, you will be presented with a graphical and visual representation of the flows and dependencies in a Directed Acyclic Graph (DAG). This helps track the flow of data from raw ingestion to the final stages of processing.&nbsp;</SPAN></P> <P><SPAN>The Protobuf deserialization occurs while populating the </SPAN><FONT face="courier new,courier"><SPAN>bronze_fleet_events</SPAN></FONT><SPAN>, which serves as a source for performing change data capture into the silver table </SPAN><FONT face="courier new,courier"><SPAN>silver_fleet_events</SPAN></FONT><SPAN>. Refer to the detailed sections below for each table to understand the data flow within this pipeline.&nbsp;</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="DLT-Pipeline-Graph.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/11416iB370985B42FD7FF4/image-size/large?v=v2&amp;px=999" role="button" title="DLT-Pipeline-Graph.png" alt="DLT-Pipeline-Graph.png" /></span></P> <P><SPAN>Leveraging the Serverless option for Delta Live Tables offers several key advantages, making it an ideal choice for efficiently processing Protobuf data without the overhead of managing infrastructure. It provides an easy-to-use and simple experience and builds a performant, auto-optimized, and fully managed data pipeline that will decrease overall TCO. You can enable Serverless DLT by navigating to the pipeline settings and ticking the "Serverless" option.&nbsp;</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="DLT-Pipeline-Settings.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/11417i1155EF18A7495AC9/image-size/large?v=v2&amp;px=999" role="button" title="DLT-Pipeline-Settings.png" alt="DLT-Pipeline-Settings.png" /></span></P> <P><SPAN>For more detailed insights on the benefits of using serverless DLT, including how it enables cost-effective, incremental ETL pipelines, refer to this latest blog: "</SPAN><A href="https://www.databricks.com/blog/cost-effective-incremental-etl-serverless-compute-delta-live-tables-pipelines" target="_blank" rel="noopener"><SPAN>Cost-effective, incremental ETL with serverless compute for Delta Live Tables pipelines</SPAN></A><SPAN>".</SPAN></P> <P>&nbsp;</P> <H1><SPAN>The 'Raw' Protobuf Event Data</SPAN></H1> <P><SPAN>Protobuf messages are often transmitted in consolidated batches rather than as individual events in high-throughput scenarios. They are prone to data repetition due to replaying past events from edge devices.&nbsp;</SPAN></P> <P><SPAN>Data is typically structured using a Protobuf schema defined in the&nbsp;</SPAN><FONT face="courier new,courier"><SPAN>.proto</SPAN></FONT><SPAN> file. An example of one is shown below. The schema includes two message types: </SPAN><FONT face="courier new,courier"><SPAN>fleetevent</SPAN></FONT><SPAN> and </SPAN><FONT face="courier new,courier"><SPAN>fleetevents</SPAN></FONT><SPAN>.</SPAN></P> <P><STRONG>fleet_management.proto</STRONG></P> <TABLE border="1" width="100%"> <TBODY> <TR> <TD width="100%" style="border-style: hidden; width: 100%;"><LI-CODE lang="python">syntax = "proto3"; package blog.fleet.tracking; message fleetevent { string vehicle_id = 1; int64 timestamp = 2; double latitude = 3; double longitude = 4; double speed = 5; double fuel_level = 6; string engine_status = 7; } message fleetevents { repeated fleetevent records = 1; }</LI-CODE></TD> </TR> </TBODY> </TABLE> <P>&nbsp;</P> <P><SPAN>To optimize data transmission, individual fleet events </SPAN><SPAN>(</SPAN><FONT face="courier new,courier"><SPAN>fleetevent</SPAN></FONT><SPAN>)&nbsp;</SPAN><SPAN>are consolidated into batches </SPAN><SPAN>(</SPAN><FONT face="courier new,courier"><SPAN>fleetevents</SPAN></FONT><SPAN>)</SPAN><SPAN>, meaning several events are combined into a single message and then serialized into the Protobuf format before sent to Kafka.&nbsp;</SPAN></P> <P><SPAN>Here is an example of a raw Protobuf payload retrieved from the Kafka topic. This payload contains a batch of multiple events, each corresponding to individual telemetry readings from a vehicle.</SPAN></P> <P>&nbsp;</P> <TABLE border="1" width="100%"> <TBODY> <TR> <TD width="100%" style="border-style: hidden; width: 100%;"><LI-CODE lang="python">"records": [ { "vehicle_id": "vehicle-27", "timestamp": "1726397972207", "latitude": 38.495878, "longitude": 139.568327, "speed": 6.95, "fuel_level": 59.42, "engine_status": "SERVICE_REQUIRED" }, ... ... ... { "vehicle_id": "vehicle-27", "timestamp": "1726397972212", "latitude": -83.874246, "longitude": 172.466784, "speed": 88.32, "fuel_level": 79.39, "engine_status": "SERVICE_REQUIRED" } ] } </LI-CODE></TD> </TR> </TBODY> </TABLE> <H2>&nbsp;</H2> <H1><SPAN>Ingesting Raw Data into Bronze Layer</SPAN></H1> <P><SPAN>The raw Protobuf data is ingested into the Bronze layer in two steps:</SPAN></P> <UL> <LI style="font-weight: 400;" aria-level="1"><FONT face="courier new,courier"><SPAN>bronze_fleet_events_raw</SPAN></FONT><SPAN> - the original Protobuf events in the binary format.</SPAN></LI> <LI style="font-weight: 400;" aria-level="1"><FONT face="courier new,courier">bronze_fleet_events</FONT><SPAN> - the original events deserialized into the structured format.</SPAN></LI> </UL> <P>&nbsp;</P> <H2><SPAN>Ingesting Protobuf Events into Databricks Incrementally</SPAN></H2> <P><SPAN>The Bronze raw streaming table is where raw fleet telemetry events are ingested from Kafka in Protobuf binary format and ready for further processing. In the DLT pipeline, raw events are ingested into the </SPAN><FONT face="courier new,courier"><SPAN>bronze_fleet_events_raw</SPAN></FONT><SPAN> using the Spark </SPAN><FONT face="courier new,courier"><SPAN>readStream</SPAN></FONT><SPAN> method. This method continuously ingests streaming data from the Kafka topic in real time.&nbsp;</SPAN></P> <P><SPAN>Streaming tables in DLT are stateful, incrementally computed, and only process data added since the last pipeline run. </SPAN><SPAN>The </SPAN><FONT face="courier new,courier"><SPAN>readStream</SPAN></FONT><SPAN> query that defines the </SPAN><FONT face="courier new,courier"><SPAN>bronze_fleet_events_raw</SPAN></FONT><SPAN> is provided in the corresponding function.</SPAN></P> <P>&nbsp;</P> <TABLE border="1" width="100%"> <TBODY> <TR> <TD width="100%" style="border-style: hidden; width: 100%;"><LI-CODE lang="python"># Bronze raw streaming table @dlt.table( name="bronze_fleet_events_raw", comment="bronze (raw) table that ingests the protobuf events from the Kafka topic" ) def bronze_fleet_events_raw(): return ( spark.readStream .format("kafka") .options(**kafka_read_options) .load() )</LI-CODE></TD> </TR> </TBODY> </TABLE> <P>&nbsp;</P> <P><STRONG>Sample result: </STRONG><SPAN>&nbsp;</SPAN><FONT face="courier new,courier"><SPAN>bronze_fleet_events_raw </SPAN></FONT><SPAN>contains the raw Protobuf messages ingested from Kafka&nbsp;in the <FONT face="courier new,courier">value</FONT> column.</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="bronze_fleet_events_raw.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/11418iD74A8DB70F919734/image-size/large?v=v2&amp;px=999" role="button" title="bronze_fleet_events_raw.png" alt="bronze_fleet_events_raw.png" /></span></P> <DIV id="tinyMceEditorsrikanthkilari_3" class="mceNonEditable lia-copypaste-placeholder">&nbsp;</DIV> <H2><SPAN>Deserializing Protobuf Binary Data</SPAN></H2> <P><SPAN>Deserialization converts the Protobuf binary data into readable structured data that can be processed and queried within the pipeline. The function definition of the Bronze table <FONT face="courier new,courier">bronze_fleet_events</FONT> consists of the Protobuf deserialization logic using the </SPAN><FONT face="courier new,courier"><SPAN>from_protobuf</SPAN></FONT><SPAN> function.&nbsp;</SPAN></P> <P><SPAN>The </SPAN><FONT face="courier new,courier"><SPAN>value</SPAN></FONT><SPAN> column (containing the payload) and </SPAN><FONT face="courier new,courier"><SPAN>schema_registry_options</SPAN></FONT><SPAN> are passed as parameters to the </SPAN><FONT face="courier new,courier"><SPAN>from_protobuf</SPAN></FONT><SPAN> function, which deserializes the payload into a struct. The </SPAN><SPAN>explode</SPAN><SPAN> function is used here to flatten the batch messages (</SPAN><FONT face="courier new,courier"><SPAN>fleetevents</SPAN></FONT><SPAN>) into the individual event (</SPAN><FONT face="courier new,courier"><SPAN>fleetevent</SPAN></FONT><SPAN>) records for further processing. Kafka metadata fields such as <FONT face="courier new,courier">key</FONT>, <FONT face="courier new,courier">partition</FONT>, <FONT face="courier new,courier">offset</FONT>, and <FONT face="courier new,courier">timestamp</FONT> are retained in this table to maintain traceability from the raw Kafka data.</SPAN></P> <P>&nbsp;</P> <TABLE border="1" width="100%"> <TBODY> <TR> <TD width="100%" style="border-style: hidden; width: 100%;"><LI-CODE lang="python"># Bronze deserialized streaming table @dlt.table( name="bronze_fleet_events", comment="bronze (deserialized) table that handles the deserialization of the protobuf using the from_protobuf function" ) def bronze_fleet_events(): return ( dlt.read_stream("bronze_fleet_events_raw") .select("key","partition","offset","timestamp", # metadata fields from_protobuf("value", options = schema_registry_options).alias("batch_message")) #payload .select("key","partition","offset","timestamp", explode("batch_message.*").alias("event"))\ .selectExpr( "event.vehicle_id", "event.timestamp as event_timestamp", "event.latitude", "event.longitude", "event.speed", "event.fuel_level", "event.engine_status", "key","partition","offset","timestamp" ) ) </LI-CODE></TD> </TR> </TBODY> </TABLE> <P>&nbsp;</P> <P><STRONG>Sample result:</STRONG> <FONT face="courier new,courier"><SPAN>bronze_fleet_events </SPAN></FONT><SPAN>contains the </SPAN><SPAN>deserialized data</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="bronze_fleet_events.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/11419i1784F98A8ACF4133/image-size/large?v=v2&amp;px=999" role="button" title="bronze_fleet_events.png" alt="bronze_fleet_events.png" /></span></P> <P><SPAN>The </SPAN><FONT face="courier new,courier"><SPAN>from_protobuf</SPAN></FONT><SPAN> function by default uses </SPAN><FONT face="courier new,courier"><SPAN>FAILFAST</SPAN></FONT><SPAN> parsing mode, which will immediately halt the pipeline and raise an error if any malformed records are encountered during the deserialization process.&nbsp;</SPAN><SPAN>Refer to the Databricks </SPAN><A href="https://docs.databricks.com/en/structured-streaming/protocol-buffers.html#supported-options-in-protobuf-functions" target="_blank" rel="noopener"><SPAN>documentation</SPAN></A><SPAN> for more details.&nbsp;</SPAN></P> <P>&nbsp;</P> <H2><SPAN>Native Databricks Protobuf Functions</SPAN></H2> <P data-unlink="true"><SPAN>To use the <A href="https://docs.databricks.com/en/structured-streaming/protocol-buffers.html#how-to-deserialize-and-serialize-protocol-buffers" target="_blank" rel="noopener">Protobuf functions</A></SPAN><FONT face="courier new,courier"><SPAN>&nbsp;from_protobuf&nbsp;</SPAN></FONT><SPAN>and </SPAN><FONT face="courier new,courier"><SPAN>to_protobuf</SPAN></FONT><SPAN>, you must provide either a schema registry via the </SPAN><SPAN>options</SPAN><SPAN> argument or a descriptor file identified by the </SPAN><SPAN>descFilePath</SPAN><SPAN> argument.&nbsp;</SPAN></P> <P><SPAN>The </SPAN><FONT face="courier new,courier"><SPAN>from_protobuf</SPAN></FONT><SPAN> and </SPAN><FONT face="courier new,courier"><SPAN>to_protobuf</SPAN></FONT><SPAN> functions are </SPAN>available on Databricks Runtime 12.2 LTS and above. <SPAN>The basic syntax for these functions is as follows:</SPAN></P> <P>&nbsp;</P> <TABLE border="1" width="100%"> <TBODY> <TR> <TD width="100%" style="border-style: hidden; width: 100%;"><LI-CODE lang="python">#Casts binary column to a struct from_protobuf( data: 'ColumnOrName' , messageName: Optional[str] = None , descFilePath: Optional[str] = None , options: Optional[Dict[str, str]] = None ) #Casts a struct back to a binary to_protobuf( data: 'ColumnOrName' , messageName: Optional[str] = None , descFilePath: Optional[str] = None , options: Optional[Dict[str, str]] = None ) </LI-CODE></TD> </TR> </TBODY> </TABLE> <P>&nbsp;</P> <P><SPAN>It is considered best practice to use a </SPAN><A href="https://docs.confluent.io/platform/current/schema-registry/index.html" target="_blank" rel="noopener"><SPAN>Schema Registry</SPAN></A><SPAN> as it ensures centralized management and versioning of schemas. This provides a robust mechanism for maintaining consistency and compatibility across producers and consumers in the data pipeline. The functions simplify the serialization and deserialization processes by leveraging schema stored in the schema registry, eliminating the need to write the custom serialization and deserialization logic manually.&nbsp;</SPAN></P> <P><SPAN>Sensitive information, such as credentials for Kafka brokers and the Schema Registry, can be securely managed using Databricks </SPAN><A href="https://docs.databricks.com/en/security/secrets/secrets.html" target="_blank" rel="noopener"><SPAN>Secrets</SPAN></A><SPAN>.</SPAN></P> <H1>&nbsp;</H1> <H1><SPAN>Cleansing for Data Integrity in Silver Layer</SPAN></H1> <P><SPAN>The </SPAN><FONT face="courier new,courier"><SPAN>silver_fleet_events</SPAN></FONT><SPAN> table handles event replays from edge devices and ensures data integrity by storing the latest event data using an SCD Type-1 approach. SCD Type-1 is used here to store only the latest version of events, ensuring that any replayed or previously sent events from edge devices are correctly updated in the table. The </SPAN><A href="https://docs.databricks.com/en/delta-live-tables/cdc.html" target="_blank" rel="noopener"><FONT face="courier new,courier"><SPAN>apply_changes</SPAN></FONT></A><SPAN> function in the DLT pipeline simplifies the change data capture (CDC) process. This function requires specifying the source, target, keys, sequencing for the change feed, and SCD-Type.&nbsp;</SPAN></P> <P><SPAN>Since edge devices can transmit previously sent events into the Kafka topic again, the Silver table </SPAN><FONT face="courier new,courier"><SPAN>silver_fleet_events</SPAN></FONT><SPAN> is configured as an SCD type-1 table using the apply_changes function. When a previously sent event is ingested again into the DLT pipeline, the corresponding entry in the Silver table gets updated. This table excludes Kafka-specific metadata columns (<FONT face="courier new,courier">key</FONT>, <FONT face="courier new,courier">partition</FONT>, <FONT face="courier new,courier">offset</FONT>, and <FONT face="courier new,courier">timestamp</FONT>) and focuses on business-relevant event data for further processing.</SPAN></P> <P>&nbsp;</P> <TABLE border="1" width="100%"> <TBODY> <TR> <TD width="100%" style="border-style: hidden; width: 100%;"><LI-CODE lang="python"># Silver streaming table dlt.create_target_table( name = "silver_fleet_events", comment = "Silver table (SCD type-1) that handles event replays from edge devices to ensure data integrity" ) dlt.apply_changes( target = "silver_fleet_events", source = "bronze_fleet_events", keys = ["vehicle_id","event_timestamp"], except_column_list = ["key","partition","offset","timestamp"], sequence_by = col("event_timestamp"), stored_as_scd_type = 1 ) </LI-CODE></TD> </TR> </TBODY> </TABLE> <P>&nbsp;</P> <P><STRONG>Sample result:</STRONG> <FONT face="courier new,courier"><SPAN>silver_fleet_events </SPAN></FONT><SPAN>is an SCD Type-1 table that ensures only the latest fleet telemetry events are stored.</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="silver_fleet_events.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/11422iA7B0181574603BDB/image-size/large?v=v2&amp;px=999" role="button" title="silver_fleet_events.png" alt="silver_fleet_events.png" /></span></P> <H1>&nbsp;</H1> <H1><SPAN>Aggregating Insights in Gold Layer</SPAN></H1> <P><SPAN>Lastly, the Silver table is used to populate two Gold summary materialized views, all within the same DLT pipeline.</SPAN></P> <UL> <LI><SPAN>The </SPAN><FONT face="courier new,courier"><SPAN>gold_fuel_efficiency</SPAN></FONT><SPAN> computes fuel efficiency as the distance traveled per unit of fuel consumed by the vehicle. The function</SPAN> <SPAN>fuel_efficiency_summary</SPAN><SPAN> contains the view's aggregation logic.</SPAN></LI> <LI><SPAN>The </SPAN><FONT face="courier new,courier"><SPAN>gold_vehicle_utilization</SPAN></FONT><SPAN> calculates the utilization rate of each vehicle by measuring how long the vehicle has been active based on its speed. The function </SPAN><SPAN>vehicle_utilisation_summary</SPAN><SPAN> contains the aggregation logic that underpins this view.&nbsp;</SPAN></LI> </UL> <TABLE border="1" width="100%"> <TBODY> <TR> <TD width="100%" style="border-style: hidden; width: 100%;"><LI-CODE lang="python"># Gold materialized views @dlt.table( name="gold_fuel_efficiency", comment="calculates fuel efficiency (distance per unit of fuel)" ) def fuel_efficiency_summary(): return ( dlt.read("silver_fleet_events") .withColumn("prev_fuel_level", lag("fuel_level").over(window_spec)) .withColumn("fuel_consumed", (col("prev_fuel_level") - col("fuel_level")).cast("double")) .withColumn("prev_latitude", lag("latitude").over(window_spec)) .withColumn("prev_longitude", lag("longitude").over(window_spec)) .withColumn("distance_traveled", when(col("prev_latitude").isNotNull(),((col("latitude") - col("prev_latitude"))**2 +(col("longitude") - col("prev_longitude"))**2)**0.5).otherwise(0)) .filter(col("fuel_consumed") &gt; 0) .groupBy("vehicle_id").agg( round(sum("distance_traveled"),2).alias("total_distance_traveled"), round(sum("fuel_consumed"),2).alias("total_fuel_consumed") ).withColumn("fuel_efficiency",round(col("total_distance_traveled") / col("total_fuel_consumed"),2)) ) @dlt.table( name="gold_vehicle_utilization", comment="calculates the vehicle utilization rate" ) def vehicle_utilisation_summary(): return ( dlt.read("silver_fleet_events") .withColumn("is_active", col("speed") &gt; 0) .groupBy("vehicle_id").agg( sum(when(col("is_active"), col("event_timestamp")).otherwise(0)).alias("total_active_time"), sum(col("event_timestamp")).alias("total_time") ).withColumn("utilization_rate", (col("total_active_time") / col("total_time")) * 100) )</LI-CODE></TD> </TR> </TBODY> </TABLE> <DIV id="tinyMceEditorsrikanthkilari_6" class="mceNonEditable lia-copypaste-placeholder">&nbsp;</DIV> <P><STRONG>Sample results:&nbsp;</STRONG></P> <UL> <LI style="font-weight: 400;" aria-level="1"><SPAN><FONT face="courier new,courier">gold_fuel_efficiency</FONT>: Aggregates total distance traveled vs. fuel consumption.</SPAN></LI> <LI style="font-weight: 400;" aria-level="1"><SPAN><FONT face="courier new,courier">gold_vehicle_utilization</FONT>: Calculates vehicle utilization rates across time.</SPAN></LI> </UL> <P><SPAN><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="gold materialized views.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/11424iD7ED88EA3CF1BD81/image-size/large?v=v2&amp;px=999" role="button" title="gold materialized views.png" alt="gold materialized views.png" /></span></SPAN></P> <H2>&nbsp;</H2> <H1><SPAN>Conclusion</SPAN></H1> <P><SPAN>Protobuf deserialization is a crucial step in processing high-volume, real-time data streams, especially when dealing with consolidated messages from edge devices. By leveraging the native from_protobuf function within a Serverless DLT pipeline, data engineering teams can simplify the deserialization process, reduce infrastructure overhead, improve deserialization performance, and minimize maintenance efforts.&nbsp;</SPAN></P> <P><SPAN>The Serverless nature of DLT further enhances this approach by eliminating the need for infrastructure management, allowing teams to focus solely on data processing and analytics. This streamlines the ingestion and processing of complex Protobuf messages and ensures that your data pipelines are scalable, cost-efficient, and easy to manage. Using declarative syntax in DLT, you can configure SCD Type 1 and 2 and other ETL needs (such as </SPAN><A href="https://docs.databricks.com/en/delta-live-tables/expectations.html" target="_blank" rel="noopener"><SPAN>Data Quality Expectations</SPAN></A><SPAN>) to build highly efficient ETL pipelines that support real-time analytics and decision-making.&nbsp;</SPAN></P> <P><SPAN>As streaming data continues to play a pivotal role across industries, adopting best practices like these in your ETL pipelines will be essential for maintaining a competitive edge and optimizing operational efficiency. Following the steps outlined in this blog, you can confidently implement Protobuf deserialization in your Serverless Databricks DLT pipelines, ensuring that your streaming applications are optimized for performance, scalability, and accuracy.</SPAN></P> Wed, 25 Sep 2024 12:47:50 GMT srikanth-kilari 2024-09-25T12:47:50Z https://community.databricks.com/t5/technical-blog/real-time-vehicle-fleet-analytics-with-databricks-delta-live/ba-p/91422 <P>Unlock real-time insights into vehicle fleet management with Databricks Delta Live Tables (DLT). This blog explores how a logistics provider leverages Databricks native functionality for Protobuf deserialization with a serverless DLT pipeline to optimize fleet performance. Discover how scalable, automated data pipelines utilize these capabilities to transform real-time telemetry data into actionable insights, improving fuel efficiency and vehicle utilization.</P> Wed, 25 Sep 2024 12:47:50 GMT https://community.databricks.com/t5/technical-blog/real-time-vehicle-fleet-analytics-with-databricks-delta-live/ba-p/91422 srikanth-kilari 2024-09-25T12:47:50Z