Important dataflow configuration

jlian · jlian · commit 290f04d4e6ae · 2024-11-01T13:33:29.000-07:00
diff --git a/articles/iot-operations/connect-to-cloud/howto-configure-dataflow-endpoint.md b/articles/iot-operations/connect-to-cloud/howto-configure-dataflow-endpoint.md
@@ -27,6 +27,24 @@ Use the following table to choose the endpoint type to configure:
 | [Microsoft Fabric OneLake](howto-configure-fabric-endpoint.md) | For uploading data to Microsoft Fabric OneLake lakehouses. | No | Yes |
 | [Local storage](howto-configure-local-storage-endpoint.md) | For sending data to a locally available persistent volume, through which you can upload data via Azure Container Storage enabled by Azure Arc edge volumes. | No | Yes |
 
+## Dataflows must use local MQTT broker endpoint
+
+When you create a dataflow, you specify the source and destination endpoints. The dataflow moves data from the source endpoint to the destination endpoint. You can use the same endpoint for multiple dataflows, and you can use the same endpoint as both the source and destination in a dataflow.
+
+However, using custom endpoints as both the source and destination in a dataflow isn't supported. This means the built-in MQTT broker in Azure IoT Operations must be either the source or destination for every dataflow. To avoid dataflow deployment failures, use the [default MQTT dataflow endpoint](./howto-configure-mqtt-endpoint.md#default-endpoint) as either the source or destination for every dataflow. 
+
+The specific requirement is that each dataflow must have either the source or destination with an MQTT endpoint with the host `aio-broker`. So while using the default endpoint isn't strictly required, and you can create new other redundant dataflow endpoints pointing to the local MQTT broker as long as the host is `aio-broker`, it's recommended to use the default endpoint to avoid confusion and manageability issues.
+
+The following table shows the supported scenarios:
+
+| Scenario | Supported |
+|----------|-----------|
+| Default endpoint as source | Yes |
+| Default endpoint as destination | Yes |
+| Custom endpoint as source | Yes, if destination is default endpoint or an MQTT endpoint with host `aio-broker` |
+| Custom endpoint as destination | Yes, if source is default endpoint or an MQTT endpoint with host `aio-broker` |
+| Custom endpoint as source and destination | No, unless one of them is an MQTT endpoints with host `aio-broker` |
+
 ## Reuse endpoints
 
 Think of each dataflow endpoint as a bundle of configuration settings that contains where the data should come from or go to (the `host` value), how to authenticate with the endpoint, and other settings like TLS configuration or batching preference. So you just need to create it once and then you can reuse it in multiple dataflows where these settings would be the same.
diff --git a/articles/iot-operations/connect-to-cloud/howto-configure-mqtt-endpoint.md b/articles/iot-operations/connect-to-cloud/howto-configure-mqtt-endpoint.md
@@ -6,7 +6,7 @@ ms.author: patricka
 ms.service: azure-iot-operations
 ms.subservice: azure-data-flows
 ms.topic: how-to
-ms.date: 10/30/2024
+ms.date: 11/01/2024
 ai-usage: ai-assisted
 
 #CustomerIntent: As an operator, I want to understand how to understand how to configure dataflow endpoints for MQTT sources and destinations in Azure IoT Operations so that I can send data to and from MQTT brokers.
@@ -25,17 +25,24 @@ MQTT dataflow endpoints are used for MQTT sources and destinations. You can conf
 
 ## Azure IoT Operations local MQTT broker
 
+Azure IoT Operations provides a [built-in local MQTT broker](../manage-mqtt-broker/overview-iot-mq.md) that you can use with dataflows. You can use the MQTT broker as a source to receive messages from other systems or as a destination to send messages to other systems.
+
 ### Default endpoint
 
-Azure IoT Operations provides a built-in MQTT broker that you can use with dataflows. When you deploy Azure IoT Operations, an MQTT broker dataflow endpoint named "default" is created with default settings. You can use this endpoint as a source or destination for dataflows. The default endpoint uses the following settings:
+When you deploy Azure IoT Operations, an MQTT broker dataflow endpoint named "default" is created with default settings. You can use this endpoint as a source or destination for dataflows. 
+
+> [!IMPORTANT]
+> The default endpoint **must always be used as either the source or destination in every dataflow**. To learn more about, see [Dataflows must use local MQTT broker endpoint](./howto-configure-dataflow-endpoint.md#dataflows-must-use-local-mqtt-broker-endpoint).
+
+The default endpoint uses the following settings:
 
 - Host: `aio-broker:18883` through the [default MQTT broker listener](../manage-mqtt-broker/howto-configure-brokerlistener.md#default-brokerlistener)
 - Authentication: service account token (SAT) through the [default BrokerAuthentication resource](../manage-mqtt-broker/howto-configure-authentication.md#default-brokerauthentication-resource)
 - TLS: Enabled
 - Trusted CA certificate: The default CA certificate `azure-iot-operations-aio-ca-trust-bundle` from the [default root CA](../deploy-iot-ops/concept-default-root-ca.md)
 
-> [!IMPORTANT]
-> If any of these default MQTT broker settings change, the dataflow endpoint must be updated to reflect the new settings. For example, if the default MQTT broker listener changes to use a different service name `my-mqtt-broker` and port 8885, you must update the endpoint to use the new host `host: my-mqtt-broker:8885`. Same applies to other settings like authentication and TLS.
+> [!CAUTION]
+> Don't delete the default endpoint. If you delete the default endpoint, you must recreate it with the same settings.
 
 To view or edit the default MQTT broker endpoint settings:
 
@@ -104,7 +111,7 @@ kubectl get dataflowendpoint default -n azure-iot-operations -o yaml
 
 ### Create new endpoint
 
-You can also create new local MQTT broker endpoints with custom settings. For example, you can create a new MQTT broker endpoint using a different port, authentication, or other settings.
+You can also create new local MQTT broker endpoints with custom settings. For example, you can create a new MQTT broker endpoint using a different port, authentication, or authorization settings. However, you must still always use the default endpoint as either the source or destination in every dataflow, even if you create new endpoints.
 
 # [Portal](#tab/portal)
 
diff --git a/articles/iot-operations/connect-to-cloud/howto-create-dataflow.md b/articles/iot-operations/connect-to-cloud/howto-create-dataflow.md
@@ -6,7 +6,7 @@ ms.author: patricka
 ms.service: azure-iot-operations
 ms.subservice: azure-data-flows
 ms.topic: how-to
-ms.date: 10/30/2024
+ms.date: 11/01/2024
 ai-usage: ai-assisted
 
 #CustomerIntent: As an operator, I want to understand how to create a dataflow to connect data sources.
@@ -39,7 +39,10 @@ flowchart LR
 
 :::image type="content" source="media/howto-create-dataflow/dataflow.svg" alt-text="Diagram of a dataflow showing flow from source to transform then destination.":::
 
-To define the source and destination, you need to configure the dataflow endpoints. The transformation is optional and can include operations like enriching the data, filtering the data, and mapping the data to another field. 
+To define the source and destination, you need to configure the dataflow endpoints. The transformation is optional and can include operations like enriching the data, filtering the data, and mapping the data to another field.
+
+> [!IMPORTANT]
+> Each dataflow must have the Azure IoT Operations local MQTT broker default endpoint [as *either* the source or destination](#proper-dataflow-configuration).
 
 You can use the operations experience in Azure IoT Operations to create a dataflow. The operations experience provides a visual interface to configure the dataflow. You can also use Bicep to create a dataflow using a Bicep template file, or use Kubernetes to create a dataflow using a YAML file.
 
@@ -168,7 +171,7 @@ Review the following sections to learn how to configure the operation types of t
 
 To configure a source for the dataflow, specify the endpoint reference and a list of data sources for the endpoint.
 
-### Use Asset as source
+### Use asset as source
 
 # [Portal](#tab/portal)
 
@@ -194,6 +197,10 @@ Configuring an asset as a source is only available in the operations experience.
 
 ---
 
+When using an asset as the source, the asset definition is used to infer the schema for the dataflow. The asset definition includes the schema for the asset's datapoints. To learn more, see [Manage asset configurations remotely](../discover-manage-assets/howto-manage-assets-remotely.md).
+
+Once configured, the data from the asset reached the dataflow via the local MQTT broker. So, when using an asset as the source, the dataflow uses the local MQTT broker default endpoint as the source in actuality.
+
 ### Use default MQTT endpoint as source
 
 # [Portal](#tab/portal)
@@ -243,7 +250,7 @@ Because `dataSources` allows you to specify MQTT or Kafka topics without modifyi
 
 ---
 
-For more information about the default MQTT endpoint and creating an MQTT endpoint as a dataflow source, see [MQTT Endpoint](howto-configure-mqtt-endpoint.md).
+If the default endpoint isn't used as the source, it must be used as the [destination](#destination). To learn more about, see [Dataflows must use local MQTT broker endpoint](./howto-configure-dataflow-endpoint.md#dataflows-must-use-local-mqtt-broker-endpoint).
 
 ### Use custom MQTT or Kafka dataflow endpoint as source
 
@@ -748,7 +755,7 @@ For more information about schema registry, see [Understand message schemas](con
 
 To configure a destination for the dataflow, specify the endpoint reference and data destination. You can specify a list of data destinations for the endpoint.
 
-To send data to a destination other than the local MQTT broker, create a dataflow endpoint. To learn how, see [Configure dataflow endpoints](howto-configure-dataflow-endpoint.md).
+To send data to a destination other than the local MQTT broker, create a dataflow endpoint. To learn how, see [Configure dataflow endpoints](howto-configure-dataflow-endpoint.md). If the destination isn't the local MQTT broker, it must be used as a source. To learn more about, see [Dataflows must use local MQTT broker endpoint](./howto-configure-dataflow-endpoint.md#dataflows-must-use-local-mqtt-broker-endpoint).
 
 > [!IMPORTANT]
 > Storage endpoints require a schema reference. If you've created storage destination endpoints for Microsoft Fabric OneLake, ADLS Gen 2, Azure Data Explorer and Local Storage, you must specify schema reference.
@@ -878,7 +885,113 @@ destinationSettings:
 
 ## Example
 
-The following example is a dataflow configuration that uses the MQTT endpoint for the source and destination. The source filters the data from the MQTT topics `thermostats/+/telemetry/temperature/#` and `humidifiers/+/telemetry/humidity/#`. The transformation converts the temperature to Fahrenheit and filters the data where the temperature is less than 100000. The destination sends the data to the MQTT topic `factory`.
+The following example is a dataflow configuration that uses the MQTT endpoint for the source and destination. The source filters the data from the MQTT topic `azure-iot-operations/data/thermostat`. The transformation converts the temperature to Fahrenheit and filters the data where the temperature multiplied by the humiditiy is less than 100000. The destination sends the data to the MQTT topic `factory`.
+
+# [Bicep](#tab/bicep)
+
+```bicep
+param aioInstanceName string = '<AIO_INSTANCE_NAME>'
+param customLocationName string = '<CUSTOM_LOCATION_NAME>'
+param dataflowName string = '<DATAFLOW_NAME>'
+
+resource aioInstance 'Microsoft.IoTOperations/instances@2024-09-15-preview' existing = {
+  name: aioInstanceName
+}
+
+resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
+  name: customLocationName
+}
+
+// Pointer to the default dataflow endpoint
+resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-09-15-preview' existing = {
+  parent: aioInstance
+  name: 'default'
+}
+
+// Pointer to the default dataflow profile
+resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-09-15-preview' existing = {
+  parent: aioInstance
+  name: 'default'
+}
+
+resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-09-15-preview' = {
+  // Reference to the parent dataflow profile, the default profile in this case
+  // Same usage as profileRef in Kubernetes YAML
+  parent: defaultDataflowProfile
+  name: dataflowName
+  extendedLocation: {
+    name: customLocation.id
+    type: 'CustomLocation'
+  }
+  properties: {
+    mode: 'Enabled'
+    operations: [
+      {
+        operationType: 'Source'
+        sourceSettings: {
+          // Use the default MQTT endpoint as the source
+          endpointRef: defaultDataflowEndpoint.name
+          // Filter the data from the MQTT topic azure-iot-operations/data/thermostat
+          dataSources: [
+            'azure-iot-operations/data/thermostat'
+          ]
+        }
+      }
+      // Transformation optional
+      {
+        operationType: 'BuiltInTransformation'
+        builtInTransformationSettings: {
+          // Filter the data where temperature * "Tag 10" < 100000
+          filter: [
+            {
+              inputs: [
+                'temperature.Value'
+                '"Tag 10".Value'
+              ]
+              expression: '$1 * $2 < 100000'
+            }
+          ]
+          map: [
+            // Passthrough all values by default
+            {
+              inputs: [
+                '*'
+              ]
+              output: '*'
+            }
+            // Convert temperature to Fahrenheit and output it to TemperatureF
+            {
+              inputs: [
+                'temperature.Value'
+              ]
+              output: 'TemperatureF'
+              expression: 'cToF($1)'
+            }
+          // Extract the "Tag 10" value and output it to Humidity
+            {
+              inputs: [
+                '"Tag 10".Value'
+              ]
+              output: 'Humidity'
+            }
+          ]
+        }
+      }
+      {
+        operationType: 'Destination'
+        destinationSettings: {
+          // Use the default MQTT endpoint as the destination
+          endpointRef: defaultDataflowEndpoint.name
+          // Send the data to the MQTT topic factory
+          dataDestination: 'factory'
+        }
+      }
+    ]
+  }
+}
+```
+
+# [Kubernetes](#tab/kubernetes)
 
 ```yaml
 apiVersion: connectivity.iotoperations.azure.com/v1beta1
@@ -887,43 +1000,52 @@ metadata:
   name: my-dataflow
   namespace: azure-iot-operations
 spec:
+  # Reference to the default dataflow profile
   profileRef: default
   mode: Enabled
   operations:
     - operationType: Source
       sourceSettings:
+        # Use the default MQTT endpoint as the source
         endpointRef: default
+        # Filter the data from the MQTT topic azure-iot-operations/data/thermostat
         dataSources:
-          - thermostats/+/telemetry/temperature/#
-          - humidifiers/+/telemetry/humidity/#
+          - azure-iot-operations/data/thermostat
+    # Transformation optional
     - operationType: builtInTransformation
       builtInTransformationSettings:
+        # Filter the data where temperature * "Tag 10" < 100000
         filter:
           - inputs:
               - 'temperature.Value'
               - '"Tag 10".Value'
-            expression: "$1*$2<100000"
+            expression: '$1 * $2 < 100000'
         map:
+          # Passthrough all values by default
           - inputs:
               - '*'
             output: '*'
+          # Convert temperature to Fahrenheit and output it to TemperatureF
           - inputs:
               - temperature.Value
             output: TemperatureF
             expression: cToF($1)
+          # Extract the "Tag 10" value and output it to Humidity
           - inputs:
               - '"Tag 10".Value'
-            output: 'Tag 10'
+            output: 'Humidity'
     - operationType: Destination
       destinationSettings:
+        # Use the default MQTT endpoint as the destination
         endpointRef: default
+        # Send the data to the MQTT topic factory
         dataDestination: factory
 ```
 
-<!-- TODO: add links to examples in the reference docs -->
-
 ---
 
+To see more examples of dataflow configurations, see [Azure REST API - Dataflow](/rest/api/iotoperations/dataflow/create-or-update#examples) and the [quickstart Bicep](https://github.com/Azure-Samples/explore-iot-operations/blob/main/samples/quickstarts/quickstart.bicep).
+
 ## Verify a dataflow is working
 
 Follow [Tutorial: Bi-directional MQTT bridge to Azure Event Grid](tutorial-mqtt-bridge.md) to verify the dataflow is working.
@@ -950,6 +1072,18 @@ kubectl get dataflow my-dataflow -o yaml > my-dataflow.yaml
 
 ---
 
+## Proper dataflow configuration
+
+To ensure the dataflow is working as expected, verify the following:
+
+- The default MQTT dataflow endpoint [must be used as *either* the source or destination](./howto-configure-dataflow-endpoint.md#dataflows-must-use-local-mqtt-broker-endpoint).
+- The dataflow profile exists and is referenced in the dataflow configuration.
+- Source is either an MQTT endpoint, Kafka endpoint, or an asset. [Storage type endpoints can't be used as a source](./howto-configure-dataflow-endpoint.md).
+- When using Event Grid as the source, the [dataflow profile instance count](./howto-configure-dataflow-profile.md#scaling) is set to 1 because Event Grid MQTT broker doesn't support shared subscriptions.
+- When using Event Hubs as the source, each event hub in the namespace is a separate Kafka topic and must be specified as the data source.
+- Transformation, if used, is configured with proper syntax, including consideration for proper [escapaing of special characters](./concept-dataflow-mapping.md#escaping).
+- When using storage type endpoints as destination, a [schema reference is specified](#serialize-data-according-to-a-schema).
+
 ## Next steps
 
 - [Map data by using dataflows](concept-dataflow-mapping.md)
