{"id":21477,"date":"2024-12-18T13:14:21","date_gmt":"2024-12-18T19:14:21","guid":{"rendered":"http:\/\/www.designandexecute.com\/designs\/?p=21477"},"modified":"2024-12-18T15:43:59","modified_gmt":"2024-12-18T21:43:59","slug":"walkthrough-of-a-databricks-real-time-data-processing-pipeline","status":"publish","type":"post","link":"https:\/\/www.designandexecute.com\/designs\/walkthrough-of-a-databricks-real-time-data-processing-pipeline\/","title":{"rendered":"Walkthrough of a Databricks Real-Time Data Processing Pipeline"},"content":{"rendered":"\n

Pipeline Overview:<\/strong><\/h3>\n\n\n\n

The goal of the pipeline was to process real-time telemetry data from industrial IoT (IIoT) sensors for predictive maintenance and anomaly detection. The data included temperature, pressure, and vibration metrics from hundreds of sensors.<\/p>\n\n\n\n

Steps:<\/strong><\/h4>\n\n\n\n
  1. \nData Ingestion:<\/strong>\n
    • Data was ingested using Apache Kafka<\/strong> topics. Each topic represented a sensor type (e.g., temperature, vibration).<\/li>
    • Kafka brokers distributed the high-velocity sensor data across multiple partitions for scalability.<\/li><\/ul>\nkafka_df = (\n spark.readStream.format(\"kafka\")\n .option(\"kafka.bootstrap.servers\", \"kafka_broker:9092\")\n .option(\"subscribe\", \"sensor-data\")\n .load()\n)\n<\/code>\n<\/li>
    • \nSchema Parsing and Transformation:<\/strong>\n
      • Raw sensor data was serialized as JSON. We parsed it into a structured format for processing.<\/li>
      • Example fields: sensor_id<\/code>, timestamp<\/code>, metric_type<\/code>, value<\/code>.<\/li><\/ul>\nfrom pyspark.sql.functions import from_json, col\nfrom pyspark.sql.types import StructType, StringType, FloatType, TimestampType\n\nschema = StructType([\n StructField(\"sensor_id\", StringType(), True),\n StructField(\"timestamp\", TimestampType(), True),\n StructField(\"metric_type\", StringType(), True),\n StructField(\"value\", FloatType(), True)\n])\n\nparsed_df = kafka_df.select(\n from_json(col(\"value\").cast(\"string\"), schema).alias(\"data\")\n).select(\"data.*\")\n<\/code>\n<\/li>
      • \nAnomaly Detection:<\/strong>\n
        • Applied custom logic to detect anomalies, e.g., flagging if the temperature exceeded a threshold.<\/li>
        • Used Spark SQL<\/strong> for time-window-based aggregations.<\/li><\/ul>\nfrom pyspark.sql.functions import window\n\nanomaly_df = parsed_df.filter(col(\"metric_type\") == \"temperature\").filter(col(\"value\") > 100)\nanomalies_with_window = (\n anomaly_df.groupBy(window(\"timestamp\", \"1 minute\"), \"sensor_id\")\n .count()\n .filter(col(\"count\") > 5) # Example threshold\n)\n<\/code>\n<\/li>
        • \nOutput Storage:<\/strong>\n
          • Anomalies were written to an Azure Data Lake<\/strong> for long-term storage and an Azure Event Hub<\/strong> for alert notifications.<\/li><\/ul>\nanomalies_with_window.writeStream \\\n .format(\"delta\") \\\n .option(\"checkpointLocation\", \"\/mnt\/checkpoints\/anomalies\") \\\n .start(\"\/mnt\/datalake\/anomalies\")\n<\/code>\n<\/li>
          • \nVisualization:<\/strong>\n
            • Data was fed into Power BI<\/strong> for anomaly trend visualization, using Azure Synapse Analytics as the integration layer.<\/li><\/ul>\n<\/li><\/ol>\n\n\n\n
              \n\n\n\n

              2. Trade-Offs: Micro-Batch vs. Continuous Processing in Spark Streaming<\/strong><\/h3>\n\n\n\n

              Micro-Batch Processing:<\/strong><\/h4>\n\n\n\n
              • How It Works:<\/strong> Data is processed in small, discrete time intervals (e.g., every second or minute).<\/li>
              • Pros:<\/strong>\n
                • Easier to implement and debug.<\/li>
                • Can use existing batch processing logic.<\/li>
                • Achieves near real-time latency with less complexity.<\/li><\/ul>\n<\/li>
                • Cons:<\/strong>\n
                  • Slight latency due to batching intervals.<\/li>
                  • Inefficient for low-latency use cases like fraud detection or industrial automation.<\/li><\/ul>\n<\/li><\/ul>\n\n\n\n

                    Continuous Processing:<\/strong><\/h4>\n\n\n\n
                    • How It Works:<\/strong> Data is processed record-by-record as it arrives.<\/li>
                    • Pros:<\/strong>\n
                      • Lowest latency, ideal for real-time use cases.<\/li>
                      • Suitable for event-driven architectures.<\/li><\/ul>\n<\/li>
                      • Cons:<\/strong>\n
                        • Harder to debug and maintain.<\/li>
                        • Requires careful resource tuning for scalability.<\/li>
                        • Supported in Spark Streaming, but not widely adopted due to implementation complexity.<\/li><\/ul>\n<\/li><\/ul>\n\n\n\n

                          Trade-Off:<\/strong><\/h4>\n\n\n\n

                          For most use cases, micro-batching<\/strong> strikes a balance between simplicity and performance. Continuous processing<\/strong> is chosen only for ultra-low-latency requirements, like stock trading or fraud detection.<\/p>\n\n\n\n

                          Here is a sample coding difference between Micro-Batch<\/strong> and Continuous Processing<\/strong> in Spark Streaming, using PySpark<\/strong> as an example.<\/p>\n\n\n\n

                          \n\n\n\n

                          2.1. Micro-Batch Processing<\/strong><\/h4>\n\n\n\n

                          Micro-batch processing divides the data into small, fixed-duration batches for processing.<\/p>\n\n\n\n

                          Code Example:<\/h4>\n\n\n\n
                          from pyspark.sql import SparkSession\nfrom pyspark.sql.functions import col\n\n# Create SparkSession\nspark = SparkSession.builder \\\n    .appName(\"MicroBatchExample\") \\\n    .getOrCreate()\n\n# Define streaming DataFrame (e.g., reading from Kafka topic)\ndf = spark.readStream \\\n    .format(\"kafka\") \\\n    .option(\"kafka.bootstrap.servers\", \"localhost:9092\") \\\n    .option(\"subscribe\", \"topic_name\") \\\n    .load()\n\n# Transformation: Process streaming data\ntransformed_df = df.selectExpr(\"CAST(key AS STRING)\", \"CAST(value AS STRING)\") \\\n    .withColumn(\"value_length\", col(\"value\").cast(\"string\").length())\n\n# Write output in micro-batches (e.g., to console or file sink)\nquery = transformed_df.writeStream \\\n    .outputMode(\"append\") \\\n    .format(\"console\") \\\n    .trigger(processingTime=\"5 seconds\") \\  # Micro-batch trigger interval\n    .start()\n\nquery.awaitTermination()\n<\/code><\/pre>\n\n\n\n
                          Key Points<\/strong>:<\/p>\n\n\n\n
                          • Uses trigger(processingTime=\"5 seconds\")<\/code> to process in fixed intervals (5 seconds).<\/li>
                          • Data is collected and processed in small, discrete batches.<\/li><\/ul>\n\n\n\n
                            \n\n\n\n
                            2.2 Continuous Processing<\/strong><\/h4>\n\n\n\n
                            Continuous processing (introduced in Spark 2.3) processes data with very low latency in a near-real-time fashion, skipping micro-batch intervals.<\/p>\n\n\n\n
                            Code Example:<\/h4>\n\n\n\n
                            from pyspark.sql import SparkSession\nfrom pyspark.sql.functions import col\n\n# Create SparkSession\nspark = SparkSession.builder \\\n    .appName(\"ContinuousProcessingExample\") \\\n    .getOrCreate()\n\n# Define streaming DataFrame (e.g., reading from Kafka topic)\ndf = spark.readStream \\\n    .format(\"kafka\") \\\n    .option(\"kafka.bootstrap.servers\", \"localhost:9092\") \\\n    .option(\"subscribe\", \"topic_name\") \\\n    .load()\n\n# Transformation: Process streaming data\ntransformed_df = df.selectExpr(\"CAST(key AS STRING)\", \"CAST(value AS STRING)\") \\\n    .withColumn(\"value_length\", col(\"value\").cast(\"string\").length())\n\n# Write output in continuous mode\nquery = transformed_df.writeStream \\\n    .outputMode(\"append\") \\\n    .format(\"console\") \\\n    .trigger(continuous=\"1 second\") \\  # Continuous processing\n    .start()\n\nquery.awaitTermination()\n<\/code><\/pre>\n\n\n\n
                            Key Points<\/strong>:<\/p>\n\n\n\n
                            • Uses trigger(continuous=\"1 second\")<\/code> for continuous streaming with a low-latency trigger.<\/li>
                            • Supports only append<\/code> output mode.<\/li>
                            • Continuous mode has lower latency but fewer supported sinks and transformations compared to micro-batch.<\/li><\/ul>\n\n\n\n
                              \n\n\n\n
                              Key Differences in Code:<\/strong><\/h3>\n\n\n\n
                              Feature<\/th>Micro-Batch<\/th>Continuous Processing<\/th><\/tr><\/thead>
                              Trigger Configuration<\/strong><\/td>trigger(processingTime=\"X seconds\")<\/code><\/td>trigger(continuous=\"X seconds\")<\/code><\/td><\/tr>
                              Latency<\/strong><\/td>Higher latency (batch intervals)<\/td>Lower latency (continuous flow)<\/td><\/tr>
                              Output Modes<\/strong><\/td>Supports append<\/code>, complete<\/code>, update<\/code><\/td>Only supports append<\/code><\/td><\/tr>
                              Use Case<\/strong><\/td>Suitable for batch-like processing<\/td>Best for low-latency real-time use<\/td><\/tr><\/tbody><\/table>\n\n\n\n
                              \n\n\n\n
                              Choose the mode depending on your latency requirements and support for transformations\/sinks.<\/p>\n\n\n\n
                              \n\n\n\n
                              3. Databricks Integration with Azure Services<\/strong><\/h3>\n\n\n\n
                              Azure Data Lake:<\/strong><\/h4>\n\n\n\n
                              • \nDatabricks seamlessly integrates with Azure Data Lake Storage Gen2<\/strong> for storing structured and unstructured data.\n<\/li>
                              • \nUsed Delta Lake<\/strong> format to ensure ACID compliance and efficient querying.\n<\/li>
                              • \nExample: Storing processed telemetry data for historical analysis.\ndelta_df.write.format(\"delta\").mode(\"append\").save(\"abfss:\/\/<container>@<storage_account>.dfs.core.windows.net\/sensor-data\")\n<\/code>\n<\/li><\/ul>\n\n\n\n
                                Azure Event Hub:<\/strong><\/h4>\n\n\n\n
                                • \nUsed for real-time alert notifications when anomalies were detected.\n<\/li>
                                • \nDatabricks wrote the alert data directly to Event Hub.\nalerts_df.writeStream \\\n    .format(\"eventhubs\") \\\n    .option(\"eventhubs.connectionString\", \"<EventHubConnectionString>\") \\\n    .start()\n<\/code>\n<\/li><\/ul>\n\n\n\n
                                  Azure Synapse Analytics:<\/strong><\/h4>\n\n\n\n
                                  • \nProcessed data was fed into Synapse for integration with Power BI dashboards.\ndata.write.format(\"com.databricks.spark.sqldw\") \\\n    .option(\"url\", \"<Synapse_SQL_Endpoint>\") \\\n    .option(\"tempDir\", \"abfss:\/\/<temp_dir>.dfs.core.windows.net\/\") \\\n    .save()\n<\/code>\n<\/li><\/ul>\n\n\n\n
                                    \n\n\n\n
                                    4. Fault Tolerance and Scalability in a Kafka-Based System<\/strong><\/h3>\n\n\n\n
                                    Fault Tolerance:<\/strong><\/h4>\n\n\n\n
                                    1.  Replication:<\/strong> 
                                      • Kafka replicates partitions across brokers, ensuring data availability during broker failures. <\/li><\/ul><\/li>
                                      •  Consumer Offsets:<\/strong> 
                                        • Stored in Kafka or an external database (e.g., ZooKeeper), consumers can resume from the last processed record. <\/li><\/ul><\/li>
                                        •  Checkpointing:<\/strong> 
                                          • Spark Streaming checkpoints data to persistent storage (e.g., Azure Blob) to recover state after failures. stream_df.writeStream \\     .format(\"delta\") \\     .option(\"checkpointLocation\", \"\/mnt\/checkpoints\/sensor-data\") \\     .start() <\/code> <\/li><\/ul><\/li><\/ol>\n\n\n\n
                                            Scalability:<\/strong><\/h4>\n\n\n\n
                                            1. Partitioning:<\/strong>\n
                                              • Distribute sensor data across Kafka partitions for parallel processing by Spark.<\/li><\/ul>\n<\/li>
                                              • Scaling Consumers:<\/strong>\n
                                                • Multiple Spark executors (or microservices) consume data from different partitions concurrently.<\/li><\/ul>\n<\/li>
                                                • Compression:<\/strong>\n
                                                  • Use Kafka compression (e.g., Snappy) to optimize storage and reduce network bandwidth.<\/li><\/ul>\n<\/li><\/ol>\n\n\n\n
                                                    \n\n\n\n
                                                    5. Real-World Example of Event-Driven Architecture<\/strong><\/h3>\n\n\n\n
                                                    Example: Predictive Maintenance in Manufacturing<\/strong><\/h4>\n\n\n\n
                                                    1. Event Source:<\/strong> 
                                                      • IIoT sensors emit real-time events (e.g., temperature, vibration) to Kafka topics. <\/li><\/ul><\/li>
                                                      • Event Processing:<\/strong> 
                                                        • Anomaly detection logic in Databricks processes these events. <\/li><\/ul><\/li>
                                                        • Trigger Actions: 
                                                          • If an anomaly is detected, the pipeline sends alerts to Azure Event Hub, which triggers: 
                                                            • A Logic App will notify the maintenance team.<\/li>
                                                            • A Power Automate<\/strong> workflow to create a ticket in a maintenance management system.  <\/li><\/ul><\/li><\/ul><\/li>
                                                            • Feedback Loop:<\/strong> 
                                                              • Sensor data, anomalies, and outcomes are fed back into machine learning models for continuous improvement. <\/li><\/ul><\/li><\/ol>\n\n\n\n
                                                                \n\n\n\n
                                                                6. How IoT Edge Enhances Event-Driven Architectures<\/strong><\/h3>\n\n\n\n
                                                                Key Benefits:<\/strong><\/h4>\n\n\n\n
                                                                1. Local Processing:<\/strong> 
                                                                  • IoT Edge processes data locally on devices, reducing latency and dependency on cloud connectivity.<\/li>
                                                                  • Example: Perform real-time anomaly detection directly on manufacturing equipment. <\/li><\/ul><\/li>
                                                                  • Event Filtering:<\/strong> 
                                                                    • Filters data before sending to the cloud, reducing bandwidth usage. <\/li><\/ul><\/li>
                                                                    • Offline Capability:<\/strong> 
                                                                      • IoT Edge stores events locally offline and syncs with the cloud when connectivity is restored. <\/li><\/ul><\/li>
                                                                      • Custom Modules:<\/strong> 
                                                                        • Developers can deploy custom Docker containers (e.g., ML models) on IoT Edge for event-driven processing. <\/li><\/ul><\/li><\/ol>\n\n\n\n
                                                                          Example:<\/strong><\/h4>\n\n\n\n
                                                                          An IoT Edge device in a factory detects abnormal vibrations using an on-device ML model. If an anomaly is found:<\/p>\n\n\n\n
                                                                          1. An alert is sent to the Kafka pipeline for cloud processing.<\/li>
                                                                          2. Local actions (e.g., stopping the machine) are triggered instantly without waiting for cloud input.<\/li><\/ol>\n","protected":false},"excerpt":{"rendered":"
                                                                            Pipeline Overview: The goal of the pipeline was to process real-time telemetry data from industrial IoT (IIoT) sensors for predictive maintenance and anomaly detection. The data included temperature, pressure, and vibration metrics from hundreds of sensors. Steps: Data Ingestion: Data was ingested using Apache Kafka topics. Each topic represented a sensor type (e.g., temperature, vibration). […]<\/p>\n","protected":false},"author":2,"featured_media":21478,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[31],"tags":[],"class_list":["post-21477","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-bi-data-warehouse"],"jetpack_featured_media_url":"https:\/\/www.designandexecute.com\/designs\/wp-content\/uploads\/2024\/12\/databricks_logor_stacked_rgb_1200px-2830717254.png","_links":{"self":[{"href":"https:\/\/www.designandexecute.com\/designs\/wp-json\/wp\/v2\/posts\/21477","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.designandexecute.com\/designs\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.designandexecute.com\/designs\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.designandexecute.com\/designs\/wp-json\/wp\/v2\/users\/2"}],"replies":[{"embeddable":true,"href":"https:\/\/www.designandexecute.com\/designs\/wp-json\/wp\/v2\/comments?post=21477"}],"version-history":[{"count":3,"href":"https:\/\/www.designandexecute.com\/designs\/wp-json\/wp\/v2\/posts\/21477\/revisions"}],"predecessor-version":[{"id":21516,"href":"https:\/\/www.designandexecute.com\/designs\/wp-json\/wp\/v2\/posts\/21477\/revisions\/21516"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/www.designandexecute.com\/designs\/wp-json\/wp\/v2\/media\/21478"}],"wp:attachment":[{"href":"https:\/\/www.designandexecute.com\/designs\/wp-json\/wp\/v2\/media?parent=21477"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.designandexecute.com\/designs\/wp-json\/wp\/v2\/categories?post=21477"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.designandexecute.com\/designs\/wp-json\/wp\/v2\/tags?post=21477"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}