💡
Fabric Eventstream in this case dual acts as endpoints to receive and send data.
The architecture consists of the following stages:
1. **Custom Endpoint as input Data Source:** Sensor data is ingested via a local custom endpoint(Generated in the local system via VS Code)
2. **Microsoft Fabric Eventstream:** The data flows through transformations to refine and enhance its usability.
3. **Derived Streams as output Data Source:** Processed data is sent to another stream and acts as an endpoint to send data.
4. **Eventhouse KQL Database:** Data is stored for further analytics and historical tracking.
## Prerequisites
* **Fabric Eventstream** configured in **Microsoft Fabric (Because we need to get the Eventhub Name and Primary Connection String to be included in the producer code)**
## Step 1: Simulating Humidity Sensor Data
A producer script simulates sensor data and sends it to a custom endpoint. This will be executed in VS Code. As a best practice, the Event Hub name and primary connection strings should be stored securely, such as in Azure Key Vault, and should not be exposed. However, for demonstration purposes, I will explain them as they are.
```python
# INSTALL azure-eventhub before running!
# pip command:
# pip install azure-eventhub
# Primary Stream
from azure.eventhub import EventHubProducerClient, EventData
from datetime import datetime, timezone
import hashlib
import random
import time
import json
# Replace the placeholders with your Event Hubs connection string and event hub name
EVENTHUB_NAME = 'es_d8ecd8c2-45c2-41e8-8f67-2d37e0c8112b'
CONNECTION_STR = 'Endpoint=sb:/'
# Configuration variables
MIN_HUMIDITY = 30.0 # Minimum humidity value
MAX_HUMIDITY = 60.0 # Maximum humidity value
COUNTRY = "Canada" # Country
CITY = "Toronto" # City
SLEEP_TIME = 5 # Sleep time before sending next event
# Example message
'''
{
"country": "Canada",
"city": "Toronto",
"timestamp": "2025-01-25T12:36:03.826769+00:00",
"humidity_readings": [
{
"sensor": "sensor_1",
"humidity": "45.34"
},
{
"sensor": "sensor_2",
"humidity": "50.21"
},
{
"sensor": "sensor_3",
"humidity": "55.69"
}
],
"event_id": "88709af29e138d8d906e009a800ebaddacd46d89d20ec91151e2ab91557170c5"
}
'''
# Create a producer client to send messages to the event hub
producer = EventHubProducerClient.from_connection_string(conn_str=CONNECTION_STR, eventhub_name=EVENTHUB_NAME)
def generate_fake_humidity(min_humidity, max_humidity):
"""Simulate a fake humidity reading within the specified range or generate null."""
return str(round(random.uniform(min_humidity, max_humidity), 2))
def get_random_sensor_readings(min_humidity, max_humidity):
"""Generate a list of humidity readings for random sensors."""
sensors = ["sensor_1", "sensor_2", "sensor_3"]
selected_sensors = random.sample(sensors, random.randint(1, len(sensors)))
return [{"sensor": sensor, "humidity": generate_fake_humidity(min_humidity, max_humidity)}
for sensor in selected_sensors]
def generate_event_id(payload):
"""Generate a SHA256 hash as a unique event ID."""
hash_object = hashlib.sha256(json.dumps(payload, sort_keys=True).encode('utf-8'))
return hash_object.hexdigest()
def get_current_timestamp():
"""Return the current timestamp in ISO 8601 format."""
return datetime.now(timezone.utc).isoformat()
try:
# Continuously generate and send fake humidity readings
while True:
# Create a batch.
event_data_batch = producer.create_batch()
# Generate random sensor readings
humidity_readings = get_random_sensor_readings(MIN_HUMIDITY, MAX_HUMIDITY)
# Create the payload
payload = {
"country": COUNTRY,
"city": CITY,
"timestamp": get_current_timestamp(),
"humidity_readings": humidity_readings
}
# Generate an event_id
payload["event_id"] = generate_event_id(payload)
# Format the message as JSON
message = json.dumps(payload)
# Add the JSON-formatted message to the batch
event_data_batch.add(EventData(message))
# Send the batch of events to the event hub
producer.send_batch(event_data_batch)
print(json.dumps(json.loads(message), indent=4))
print(event_data_batch)
# Wait for a bit before sending the next reading
time.sleep(SLEEP_TIME)
except KeyboardInterrupt:
print("Stopped by the user")
except Exception as e:
print(f"Error: {e}")
finally:
# Close the producer
producer.close()
```
> Here’s a brief breakdown of the code snippet:
>
> * It generates random humidity readings between 30% and 60% for three sensors.
>
> * The data includes location (`Canada, Toronto`) and a timestamp.
>
> * The payload is serialized into JSON and assigned a unique event ID using SHA-256 hashing.
>
> * The script continuously generates and prints this data every 5 seconds, simulating real-time sensor readings.
>
> * It runs in an infinite loop until manually stopped (`KeyboardInterrupt`).
>
> * Event hub name and Connection String-Primary key should be obtained from folloiwng (That’s why we need to first create an Eventstream and then select the source as CustomEndpoint )
>
> * ![](https://cdn.hashnode.com/res/hashnode/image/upload/v1738942204045/02dccf4c-a0ae-4344-820d-0673ac84df7d.png align="center")
>
Following are the payload output in VS Code
![](https://cdn.hashnode.com/res/hashnode/image/upload/v1738933296778/048e7714-04e0-4a4c-bae9-db0eac86f468.png align="center")
![](https://cdn.hashnode.com/res/hashnode/image/upload/v1738942482132/543f62eb-a36a-49b7-a3dd-f7a7f091d1d0.png align="center")
## Step 2: Transforming Data in Microsoft Fabric Eventstream
In these steps, i will explain different transformations done to the generated Eventstream output.
### **2.1 Expanding Humidity Readings**
This transformation expands the `humidity_readings` array into individual events using