你当前正在访问 Microsoft Azure Global Edition 技术文档网站。如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

配置数据流终结点

2025-06-21

重要说明

本页包含使用 Kubernetes 部署清单（目前为预览版）管理 Azure IoT 操作组件的说明。此功能提供多个限制，不应用于生产工作负荷。

有关适用于 Beta 版、预览版或尚未正式发布的 Azure 功能的法律条款，请参阅适用于 Microsoft azure 预览版的补充使用条款。

若要开始使用数据流，请先创建数据流终结点。数据流终结点是数据流的连接点。可以将终结点用作数据流的源或目标。某些终结点类型既可用作源，又可用作目标，而其他终结点类型则只能用作目标。数据流至少需要一个源终结点和一个目标终结点。

使用下表选择要配置的终结点类型：

终结点类型	说明	可用作源	可用作目标
MQTT	适用于使用 MQTT 代理的双向消息传送，包括内置到 Azure IoT 操作和事件网格的消息传送。	是	是
Kafka	适用于使用 Kafka 代理（包括 Azure 事件中心）的双向消息传送。	是	是
Data Lake	适用将数据上传到 Azure Data Lake Gen2 存储帐户。	否	是
Microsoft Fabric OneLake	将数据上传到 Microsoft Fabric OneLake 湖屋。	否	是
Azure 数据资源管理器	将数据上传到 Azure 数据资源管理器数据库。	否	是
本地存储	若要将数据发送到本地可用的永久性卷，可选择性地使用 Azure Arc 启用的 Azure 容器存储进行配置。	否	是

重要说明

存储终结点需要用于序列化的架构。若要将数据流与 Microsoft Fabric OneLake、Azure Data Lake Storage、Azure 数据资源管理器或本地存储配合使用，必须指定架构引用。

若要从示例数据文件生成架构，请使用 Schema Gen Helper。

数据流必须使用本地 MQTT 代理终结点

创建数据流时，需要指定源终结点和目标终结点。数据流将数据从源终结点移动到目标终结点。可以对多个数据流使用相同的终结点，并且可以使用与数据流中的源终结点和目标终结点相同的终结点。

但是，不支持将自定义终结点用作数据流中的源终结点和目标终结点。此限制意味着 Azure IoT 操作中的内置 MQTT 代理必须至少为一个终结点。它可以是源、目标，也可以是两者。为了避免数据流部署失败，请使用默认 MQTT 数据流终结点作为每个数据流的源或目标。

具体要求是，每个数据流都必须将源终结点或目标终结点配置为具有主机 aio-broker 的 MQTT 终结点。因此，使用默认终结点并非严格要求，只要主机是 aio-broker，就可以创建指向本地 MQTT 代理的其他数据流终结点。但是，为了避免混淆和可管理性问题，建议使用默认终结点。

下表显示了支持的方案：

场景	支持
默认终结点作为源	是
默认终结点作为目标	是
自定义终结点作为源	是，如果目标终结点是默认终结点或具有主机 `aio-broker` 的 MQTT 终结点
自定义终结点作为目标	是，如果源终结点是默认终结点或具有主机 `aio-broker` 的 MQTT 终结点
自定义终结点作为源和目标	否，除非其中一个终结点是具有主机 `aio-broker` 的 MQTT 终结点

重复使用终结点

将每个数据流终结点视为一组配置设置，其中包含数据应来自或转到的位置（host 值）、如何使用终结点进行身份验证，以及 TLS 配置或批处理首选项等其他设置。因此，只需创建一次，然后就可以在多个数据流中重复使用，其中这些设置相同。

为了更轻松地重复使用终结点，MQTT 或 Kafka 主题筛选器不属于终结点配置。改为在数据流配置中指定主题筛选器。这意味着，可以对使用不同主题筛选器的多个数据流使用相同的终结点。

例如，可以使用默认 MQTT 中转站数据流终结点。可以将它用于具有不同主题筛选器的源和目标：

使用操作经验创建从 MQTT 到 MQTT 的数据流的屏幕截图。

使用 az iot ops dataflow apply 命令来创建或更改数据流。

az iot ops dataflow apply --resource-group <ResourceGroupName> --instance <AioInstanceName> --profile <DataflowProfileName> --name <DataflowName> --config-file <ConfigFilePathAndName>

参数 --config-file 是包含资源属性的 JSON 配置文件的路径和文件名。

在此示例中，假定一个配置文件 data-flow.json ，其中包含存储在用户主目录中的以下内容：

{
  "mode": "Enabled",
  "operations": [
    {
      "operationType": "Source",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "example/topic/1"
        ]
      }
    },
    {
      "operationType": "Destination",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "example/topic/2"
      }
    }
  ]
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-01' = {
  parent: <DEFAULT_PROFILE_RESOURCE>
  name: 'broker-to-broker'
  extendedLocation: {
    name: <CUSTOM_LOCATION_RESOURCE>.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    operations: [
      {
        operationType: 'Source'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'example/topic/1'
          ]
        }
      }
      {
        operationType: 'Destination'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'example/topic/2'
        }
      }
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: Dataflow
metadata:
  name: broker-to-broker
  namespace: azure-iot-operations
spec:
  profileRef: default
  operations:
    - operationType: Source
      sourceSettings:
        endpointRef: default
        dataSources:
        - example/topic/1
    - operationType: Destination
      destinationSettings:
        endpointRef: default
        dataDestination: example/topic/2

同样，可以创建多个数据流，这些数据流对其他终结点和主题使用相同的 MQTT 终结点。例如，可以使用用于将数据发送到事件中心终结点的数据流相同的 MQTT 终结点。

截屏展示利用操作经验创建从 MQTT 到 Kafka 的数据流。

使用 az iot ops dataflow apply 命令来创建或更改数据流。

az iot ops dataflow apply --resource-group <ResourceGroupName> --instance <AioInstanceName> --profile <DataflowProfileName> --name <DataflowName> --config-file <ConfigFilePathAndName>

参数 --config-file 是包含资源属性的 JSON 配置文件的路径和文件名。

在此示例中，假定一个配置文件 data-flow.json ，其中包含存储在用户主目录中的以下内容：

{
  "mode": "Enabled",
  "operations": [
    {
      "operationType": "Source",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "example/topic/3"
        ]
      }
    },
    {
      "operationType": "Destination",
      "destinationSettings": {
        // The endpoint needs to be created before you can reference it here
        "endpointRef": "example-event-hub-endpoint",
        "dataDestination": "example/topic/4"
      }
    }
  ]
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-01' = {
  parent: <DEFAULT_PROFILE_RESOURCE>
  name: 'broker-to-eh'
  extendedLocation: {
    name: <CUSTOM_LOCATION_RESOURCE>.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    operations: [
      {
        operationType: 'Source'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'example/topic/3'
          ]
        }
      }
      {
        operationType: 'Destination'
        destinationSettings: {
          // The endpoint needs to be created before you can reference it here
          endpointRef: 'example-event-hub-endpoint'
          dataDestination: 'example/topic/4'
        }
      }
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: Dataflow
metadata:
  name: broker-to-eh
  namespace: azure-iot-operations
spec:
  profileRef: default
  operations:
    - operationType: Source
      sourceSettings:
        endpointRef: default
        dataSources:
        - example/topic/3
    - operationType: Destination
      destinationSettings:
        # The endpoint needs to be created before you can reference it here
        endpointRef: example-event-hub-endpoint
        dataDestination: example/topic/4

与 MQTT 示例类似，可以创建多个数据流，这些数据流对不同的主题使用相同的 Kafka 终结点，或为不同的表使用相同的 Data Lake 终结点。

后续步骤

创建数据流终结点：

通过

配置数据流终结点

数据流必须使用本地 MQTT 代理终结点

重复使用终结点

后续步骤

反馈

其他资源