重要事項
本頁包含使用 Kubernetes 部署清單管理 Azure IoT Operations 元件的說明，該清單已在預覽中。 此功能有多項限制，不應用於生產工作負載。
請參閱Microsoft Azure 預覽版的補充使用條款，以了解適用於 Azure 測試版、預覽版或其他尚未普遍公開發佈功能的法律條款。
data flow 是資料從來源到目的地的路徑，並可選擇性轉換。 你可以透過建立 Data flow 自訂資源或使用 Operations Experience 網頁介面來配置 Data flow。 一個資料流由三個部分組成：來源、轉換，以及目標。
要定義來源和目的地，你需要設定 data flow 端點。 轉換是選擇性的，可以包含擴充資料、篩選資料，以及將資料對應至另一個欄位等作業。
重要事項
每個數據流必須擁有Azure IoT Operations的本地MQTT代理預設端點作為或是來源或目的地。
你可以在 Azure IoT Operations 中利用操作體驗來建立資料流。 操作體驗提供一個視覺化介面來配置資料流。 你可以使用 Bicep 檔案來建立資料流程，也可以使用 Kubernetes 搭配 YAML 檔案來建立資料流程。
繼續閱讀以了解如何設定來源、轉換和目的地。
先決條件
只要你有 Azure IoT Operations 實例，就可以使用預設的 data flow 配置檔和端點部署資料流。 不過，你可能想設定資料流配置文件和端點，以便自訂資料流。
Data flow 剖面
如果你不需要不同的資料流縮放設定，可以使用 Azure IoT Operations 提供的 預設資料流設定檔。 避免將過多資料流與單一資料流配置檔關聯。 如果你有大量的資料流，請將它們分散到多個資料流配置檔案，以降低超過資料流配置檔案大小限制70MB的風險。
欲了解如何設定新的data flow設定檔，請參閱 配置data flow設定檔。
資料流端點
你需要 data flow 端點來設定 data flow 的來源和目的地。 為了快速開始，請使用本地 MQTT 經紀商的預設資料流端點。 你也可以建立其他類型的 data flow 端點，例如 Kafka、Event Hubs、OpenTelemetry 或 Azure Data Lake Storage。 欲了解更多資訊，請參閱 設定資料流端點。
開始使用
當你具備必要的條件後，就可以開始建立資料流。
要在 operations experience 中建立資料流，請選擇 資料流>建立資料流。
選擇佔位符名稱 new-data-flow 來設定 data flow 屬性。 輸入 data flow 名稱，並選擇要使用的 data flow 設定檔。 預設的資料流配置檔會自動選擇。 如需有關資料流設定檔的更多資訊，請參閱 Configure data flow profile。
重要事項
你只能在建立資料流程時選擇資料流程設定檔。 在建立資料流配置檔後，您無法更改它。
如果你想更改現有資料流程的資料流程設定檔，請刪除原本的資料流程，然後使用新的資料流程設定檔建立新的資料流程。
透過選擇 data flow 圖中的項目，設定 data flow 的來源端點、轉換端點和目的端點。
使用 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": {
// See source configuration section
}
},
{
"operationType": "BuiltInTransformation",
"builtInTransformationSettings": {
// See transformation configuration section
}
},
{
"operationType": "Destination",
"destinationSettings": {
// See destination configuration section
}
}
]
}
這裡有一個範例指令，來建立或更新使用預設資料流程配置檔的資料流：
az iot ops dataflow apply --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --config-file ~/data-flow.json
建立一個 Bicep .bicep 檔案來開始建立資料流程。 此範例展示了包含來源、轉換與目的配置的 data flow 結構。
param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowName string = '<DATAFLOW_NAME>'
resource aioInstance 'Microsoft.IoTOperations/instances@2024-11-01' existing = {
name: aioInstanceName
}
resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
name: customLocationName
}
resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' existing = {
parent: aioInstance
name: 'default'
}
// Pointer to the default data flow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-11-01' existing = {
parent: aioInstance
name: 'default'
}
resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-01' = {
// Reference to the parent data flow 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: {
// See source configuration section
}
}
// Transformation optional
{
operationType: 'BuiltInTransformation'
builtInTransformationSettings: {
// See transformation configuration section
}
}
{
operationType: 'Destination'
destinationSettings: {
// See destination configuration section
}
}
]
}
}
開始建立資料流，並建立一個 Kubernetes 清單 .yaml 檔案。 此範例展示了包含來源、轉換與目的配置的 data flow 結構。
apiVersion: connectivity.iotoperations.azure.com/v1
kind: Dataflow
metadata:
name: <DATAFLOW_NAME>
namespace: azure-iot-operations
spec:
# Reference to the default data flow profile
# This field is required when configuring via Kubernetes YAML
# The syntax is different when using Bicep
profileRef: default
mode: Enabled
operations:
- operationType: Source
sourceSettings:
# See source configuration section
# Transformation optional
- operationType: BuiltInTransformation
builtInTransformationSettings:
# See transformation configuration section
- operationType: Destination
destinationSettings:
# See destination configuration section
請複習以下章節，了解如何配置 data flow 的操作類型。
來源
要配置 data flow 的來源，請指定端點參考及該端點的資料來源清單。 請選擇以下選項之一作為 data flow 的來源。
如果你不使用預設端點作為來源，就必須用它作為 目的地。 欲了解更多關於使用本地 MQTT 代理端點的資訊，請參閱資料 流必須使用本地 MQTT 代理端點。
選項 1：使用預設訊息代理程式端點作為來源
在 [來源詳細資料] 下，選取 [訊息代理程式]。
針對訊息代理程式來源輸入以下設定：
| 設定 |
描述 |
| 數據流端點 |
選取 [預設] 以使用預設 MQTT 訊息代理程式端點。 |
| 主題 |
用於訂閱傳入訊息的主題篩選條件。 使用 [主題]> [新增資料列] 來新增多個主題。 如需主題的詳細資訊，請參閱設定 MQTT 或 Kafka 主題。 |
| 訊息結構描述 |
用於反序列化傳入訊息的結構描述。 請參閱指定結構描述來還原序列化資料。 |
選取 [套用]。
以下是預設 MQTT 訊息代理程式端點的範例來源端點組態：
{
"operationType": "Source",
"sourceSettings": {
"endpointRef": "default",
"dataSources": [
"thermostats/+/sensor/temperature/#",
"humidifiers/+/sensor/humidity/#"
],
"endpointRef": "default"
}
}
你可以在 Bicep 檔案中設定訊息代理端點。 例如，以下端點是 data flow 的來源：
sourceSettings: {
endpointRef: 'default'
dataSources: [
'thermostats/+/sensor/temperature/#'
'humidifiers/+/sensor/humidity/#'
]
}
若要配置使用訊息代理端點及兩個主題過濾器的來源，請使用以下配置：
sourceSettings:
endpointRef: default
dataSources:
- thermostats/+/sensor/temperature/#
- humidifiers/+/sensor/humidity/#
因為 dataSources 可以接受 MQTT 或 Kafka 主題而不修改端點設定，即使主題不同，你也可以重複使用該端點來處理多個資料流。 欲了解更多資訊，請參閱 「設定資料來源」。
選項 2：使用資產作為來源
你可以用 asset 作為 data flow 的來源。 您只能在營運經驗中使用資產作為資源。
在 [來源詳細資料]下，選取 [資產]。
選取您要用作來源端點的資產。
選取 [繼續]。
即會顯示所選取資產的資料點清單。
選取 [套用] 以使用資產作為來源端點。
當你使用資產作為來源時，資產定義會提供 data flow 的結構。 資產定義包含資產資料點的結構描述。 欲了解更多資訊，請參閱 遠端管理資產配置。
設定來源後，資產的資料會透過本地的 MQTT 代理程式傳送到資料流程。 所以，當你用資產作為來源時，data flow 會使用本地 MQTT 代理的預設端點作為來源。
選項三：使用自訂的 MQTT 或 Kafka data flow 端點作為來源
如果你建立了自訂的 MQTT 或 Kafka data flow 端點（例如，要搭配 Event Grid 或 Event Hubs 使用），你可以用它作為 data flow 的來源。 請記得，像 Data Lake 或 Fabric OneLake 這類 storage 類型的端點，不能被用作原始碼。
在 [來源詳細資料] 下，選取 [訊息代理程式]。
針對訊息代理程式來源輸入以下設定：
選取 [套用]。
將預留位置值取代為您的自訂端點名稱和主題。
{
"operationType": "Source",
"sourceSettings": {
"endpointRef": "<CUSTOM_ENDPOINT_NAME>",
"dataSources": [
"<TOPIC_1>",
"<TOPIC_2>"
]
}
}
將預留位置值取代為您的自訂端點名稱和主題。
sourceSettings: {
endpointRef: '<CUSTOM_ENDPOINT_NAME>'
dataSources: [
'<TOPIC_1>'
'<TOPIC_2>'
// See section on configuring MQTT or Kafka topics for more information
]
}
將預留位置值取代為您的自訂端點名稱和主題。
sourceSettings:
endpointRef: <CUSTOM_ENDPOINT_NAME>
dataSources:
- <TOPIC_1>
- <TOPIC_2>
# See section on configuring MQTT or Kafka topics for more information
你可以在一個來源中指定多個 MQTT 或 Kafka 主題，而不需要修改 data flow 端點的設定。 這種彈性意味著即使主題不同，也能在多個資料流中重複使用同一個端點。 更多資訊請參見 Reuse data flow endpoints。
MQTT 主題
當來源端點是 MQTT（包含事件網格）時，使用 MQTT 主題過濾器訂閱收到的訊息。 主題篩選條件可以包含萬用字元來訂閱多個主題。 例如， thermostats/+/sensor/temperature/# 訂閱控溫器的所有溫度感測器訊息。 若要設定 MQTT 主題篩選條件：
在操作體驗資料流Source details中，選擇Message broker，然後使用Topic欄位指定訂閱 MQTT 主題篩選器以接收訊息。 要新增多個 MQTT 主題，請選擇 新增列 並輸入新主題。
{
"operationType": "Source",
"sourceSettings": {
"endpointRef": "<MESSAGE_BROKER_ENDPOINT_NAME>",
"dataSources": [
"<TOPIC_FILTER_1>",
"<TOPIC_FILTER_2>"
// Add more topic filters as needed
]
}
}
使用萬用字元的多個 MQTT 主題篩選條件的範例：
{
"operationType": "Source",
"sourceSettings": {
"endpointRef": "default",
"dataSources": [
"thermostats/+/sensor/temperature/#",
"humidifiers/+/sensor/humidity/#"
]
}
}
在此，通配符+會選取在thermostats和humidifiers主題下的所有裝置。
#萬用字元會選取temperature和humidity主題之所有子主題底下的所有感測器訊息。
sourceSettings: {
endpointRef: '<MESSAGE_BROKER_ENDPOINT_NAME>'
dataSources: [
'<TOPIC_FILTER_1>'
'<TOPIC_FILTER_2>'
// Add more topic filters as needed
]
}
使用萬用字元的多個 MQTT 主題篩選條件的範例：
sourceSettings: {
endpointRef: 'default'
dataSources: [
'thermostats/+/sensor/temperature/#'
'humidifiers/+/sensor/humidity/#'
]
}
在此，通配符+會選取在thermostats和humidifiers主題下的所有裝置。
#萬用字元會選取temperature和humidity主題之所有子主題底下的所有感測器訊息。
sourceSettings:
endpointRef: <ENDPOINT_NAME>
dataSources:
- <TOPIC_FILTER_1>
- <TOPIC_FILTER_2>
# Add more topic filters as needed
使用萬用字元的多個主題篩選條件的範例：
sourceSettings:
endpointRef: default
dataSources:
- thermostats/+/sensor/temperature/#
- humidifiers/+/sensor/humidity/#
在此，通配符+會選取在thermostats和humidifiers主題下的所有裝置。
#萬用字元會選取temperature和humidity主題之所有子主題底下的所有訊息。
共用訂閱
若要使用訊息代理來源的共享訂閱，請以 的形式指定共享訂閱主題 $shared/<GROUP_NAME>/<TOPIC_FILTER>。
在 Operations Experience 資料流程 來源詳細資訊 中，選擇 訊息中介軟體，並使用 主題 欄位來指定共用的訂閱群組和主題。
{
"operationType": "Source",
"sourceSettings": {
"dataSources": [
"$shared/<GROUP_NAME>/<TOPIC_FILTER>"
]
}
}
sourceSettings: {
dataSources: [
'$shared/<GROUP_NAME>/<TOPIC_FILTER>'
]
}
sourceSettings:
dataSources:
- $shared/<GROUP_NAME>/<TOPIC_FILTER>
如果 data flow 設定檔 的實例數超過一，所有使用訊息代理來源的資料流都會自動啟用共享訂閱。 在此情況下，會新增 $shared 前置詞並自動產生共用訂閱群組名稱。 舉例來說，如果你有一個實例數為 3 的 data flow 設定檔，且你的data flow使用訊息代理端點作為來源，並設定了主題 topic1 和 topic2，這些主題會自動轉換成共享訂閱，分別是 $shared/<GENERATED_GROUP_NAME>/topic1 和 $shared/<GENERATED_GROUP_NAME>/topic2。
您可以在組態中明確建立名為 $shared/mygroup/topic 的主題。 不過，不建議明確新增 $shared 主題，因為 $shared 前綴會在需要時自動新增。 資料流程可以使用群組名稱進行最佳化 (如果它未設定的話)。 例如，未設定 $share，且資料流程僅需對主題名稱進行運作。
重要事項
當實例數超過一且你使用 Event Grid MQTT 代理作為來源時，共享訂閱對於資料流非常重要，因為它 不支援共享訂閱。 為避免訊息遺失，使用事件網格 MQTT 代理作為來源時，將 data flow profile 的實例計數設為一。 也就是說，當data flow作為訂閱者，並接收來自雲端的訊息時。
Kafka 主題
當來源是 Kafka (包含事件中樞) 端點時，請指定要訂閱傳入訊息的個別 Kafka 主題。 不支援通配符，因此您必須以靜態方式指定每個主題。
附註
透過 Kafka 端點使用事件中樞時，命名空間內的每個個別事件中樞都是 Kafka 主題。 如果您有一個包含兩個事件中樞 (thermostats 和 humidifiers) 的事件中樞命名空間，則可以將每個事件中樞指定為一個 Kafka 主題。
若要設定 Kafka 主題：
在操作體驗資料流的Source details中，選擇Message broker，然後使用Topic欄位來指定 Kafka 主題篩選器，以便訂閱和接收新訊息。
附註
在操作體驗中，你只能指定一個主題過濾器。 若要使用多個主題篩選條件，請使用 Bicep 或 Kubernetes。
{
"operationType": "Source",
"sourceSettings": {
"endpointRef": "<KAFKA_ENDPOINT_NAME>",
"dataSources": [
"<KAFKA_TOPIC_1>",
"<KAFKA_TOPIC_2>"
// Add more Kafka topics as needed
]
}
}
sourceSettings: {
endpointRef: '<KAFKA_ENDPOINT_NAME>'
dataSources: [
'<KAFKA_TOPIC_1>'
'<KAFKA_TOPIC_2>'
// Add more Kafka topics as needed
]
}
sourceSettings:
endpointRef: <KAFKA_ENDPOINT_NAME>
dataSources:
- <KAFKA_TOPIC_1>
- <KAFKA_TOPIC_2>
# Add more Kafka topics as needed
指定來源結構描述
當你使用 MQTT 或 Kafka 作為來源時，你可以指定 一個結構 來顯示 Operations Experience 網頁介面中資料點的清單。
目前不支援使用結構描述來還原序列化和驗證傳入訊息。
如果來源是資產，入口網站會自動根據資產定義推斷出結構。
若要設定用於還原序列化來自來源的傳入訊息的結構描述：
在 Operations 體驗數據流的 Source details 中，選擇 Message broker，並使用 Message schema 欄位來指定結構描述。 請選取 上傳 按鈕以傳送結構檔案。 欲了解更多資訊，請參閱 「了解訊息結構」。
{
"operationType": "Source",
"sourceSettings": {
"endpointRef": "<ENDPOINT_NAME>",
"serializationFormat": "Json",
"schemaRef": "aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>"
}
}
在你使用 schema 登錄檔儲存 schema 後，在 data flow 配置中引用它。
sourceSettings: {
serializationFormat: 'Json'
schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>'
}
在你使用 schema 登錄檔儲存 schema 後，在 data flow 配置中引用它。
sourceSettings:
serializationFormat: Json
schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>'
欲了解更多資訊，請參閱 「了解訊息結構」。
要求磁碟持續性
要求磁碟持續性可讓數據流在重新啟動時保持狀態。 當您啟用這項功能時，如果連線的訊息代理程式重新啟動，圖形就會復原處理狀態。 此功能對於遺失中繼資料會成為問題的具狀態處理案例很有用。 當您啟用要求磁碟持續性時，訊息代理程式會將 MQTT 數據，例如訂閱者佇列中的訊息保存到磁碟。 這種做法能確保你的資料流資料來源在停電或重啟代理時不會遺失資料。 代理者能維持最佳效能，因為持久性是依 data flow 設定，只有需要持久化的資料流才會使用此功能。
資料流圖在訂閱期間使用 MQTTv5 使用者屬性來請求此持久性。 這項功能只有在下列情況下才能運作：
- 資料流使用 MQTT 代理或資產作為來源
- MQTT 訊息代理程式已啟用持續性，且動態持續性模式已針對數據類型設定為
Enabled ，例如訂閱者佇列
此設定可讓資料流程之類的 MQTT 用戶端，透過使用 MQTTv5 使用者屬性來要求其訂用帳戶的磁碟持續性。 如需 MQTT 訊息代理程式持續性設定的詳細資訊，請參閱 設定 MQTT 訊息代理程式持續性。
設定接受 Enabled 或 Disabled。
Disabled 是預設值。
建立或編輯資料流時，選擇 Edit，然後在 Request data persistence 旁邊選擇 Yes。
在你的 data flow 設定檔中加入 requestDiskPersistence 屬性：
{
"mode": "Enabled",
"requestDiskPersistence": "Enabled",
"operations": [
// ... your data flow operations
]
}
將 requestDiskPersistence 屬性加入你的 data flow 資源中。 API 版本為 2025-10-01 或更新版本：
resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2025-10-01' = {
parent: defaultDataflowProfile
name: dataflowName
extendedLocation: {
name: customLocation.id
type: 'CustomLocation'
}
properties: {
mode: 'Enabled'
requestDiskPersistence: 'Enabled'
operations: [
// ... your data flow operations
]
}
}
在你的資料流程規範中加入 requestDiskPersistence 屬性。使用 API 版本 connectivity.iotoperations.azure.com/v1beta1 或更新版本：
apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
name: <DATAFLOW_NAME>
namespace: azure-iot-operations
spec:
profileRef: default
mode: Enabled
requestDiskPersistence: Enabled
operations:
# ... your data flow operations
轉換操作是指你在將資料傳送到目的地之前，先從來源轉換資料。 轉換是選用的。 如果你不需要對資料做修改，就不要在 data flow 配置中包含轉換操作。 多個變換會分階段串連，無論你在配置中指定順序如何。 階段的順序一律是：
-
豐富：給定資料集與條件，為來源資料加入更多資料以匹配。
-
篩選：根據條件篩選資料。
-
對應、計算、重新命名或新增新屬性：使用選用轉換將資料從一個欄位移至另一個欄位。
本節介紹 data flow 轉換。 欲了解更多詳細資訊，請參閱 使用 data flows 映射資料、使用 data flow conversions 轉換資料，以及 使用 data flows 豐富資料。
在操作體驗中，選擇 Data flow>Add transform（可選）。
{
"operationType": "BuiltInTransformation",
"builtInTransformationSettings": {
"datasets": [
// See section on enriching data
],
"filter": [
// See section on filtering data
],
"map": [
// See section on mapping data
]
}
}
builtInTransformationSettings: {
datasets: [
// See section on enriching data
]
filter: [
// See section on filtering data
]
map: [
// See section on mapping data
]
}
builtInTransformationSettings:
datasets:
# See section on enriching data
filter:
# See section on filtering data
map:
# See section on mapping data
擴充：新增參考資料
為了豐富資料，首先在 Azure IoT Operations state store 中加入參考資料集。 資料集會根據條件向來源資料添加額外資料。 條件會指定為來源資料中的欄位，該欄位符合資料集中的欄位。
你可以使用 state store CLI 將樣本資料載入狀態儲存。 狀態儲存中的金鑰名稱對應於 data flow 配置中的資料集。
要豐富資料，請在data flow配置中使用 builtInTransformationSettings 屬性。 利用datasets屬性來指定資料集以進行豐富化。
{
"operationType": "BuiltInTransformation",
"builtInTransformationSettings": {
"datasets": [
{
"key": "<DATASET_KEY>",
"inputs": [
"$source.<SOURCE_FIELD>" // ---------------- $1
"$context(<DATASET_KEY>).<DATASET_FIELD>" // - $2
],
"expression": "$1 == $2"
}
]
}
}
這個範例展示了如何利用 deviceId 來源資料中的欄位來匹配 asset 資料集中的欄位：
builtInTransformationSettings: {
datasets: [
{
key: 'assetDataset'
inputs: [
'$source.deviceId' // ---------------- $1
'$context(assetDataset).asset' // ---- $2
]
expression: '$1 == $2'
}
]
}
例如，您可使用來源資料中的 deviceId 欄位來比對資料集中的 asset 欄位：
builtInTransformationSettings:
datasets:
- key: assetDataset
inputs:
- $source.deviceId # ------------- $1
- $context(assetDataset).asset # - $2
expression: $1 == $2
如果資料集具有 asset 欄位的記錄，類似於：
{
"asset": "thermostat1",
"location": "room1",
"manufacturer": "Contoso"
}
來自來源且 deviceId 欄位與 thermostat1 相符的資料在篩選和對應階段中具有可用的 location 和 manufacturer 欄位。
如需條件語法的詳細資訊，請參閱使用資料流程來擴充資料和使用資料流程來轉換資料。
篩選：根據條件篩選資料
要依條件過濾資料，請使用filter階段。 將條件指定為來源資料中的欄位，與某個值相符。
在 [轉換 (選用)] 下，選取 [篩選]>[新增]。
輸入必要的設定。
| 設定 |
描述 |
| 篩選條件 |
根據來源資料中的欄位來篩選資料的條件。 |
| 描述 |
提供篩選條件的描述。 |
在篩選條件欄位中，輸入 @ 或選取 [Ctrl + 空格鍵] 以從下拉式清單中選擇資料點。
透過格式 @$metadata.user_properties.<property> 或 @$metadata.topic輸入 MQTT 元資料屬性。 請使用 @$metadata.<header>格式 輸入$metadata標頭。 只有屬於訊息標頭一部分的 MQTT 屬性才需要 $metadata 語法。 如需詳細資訊，請參閱欄位參考。
條件可以使用來源資料中的欄位。 例如，使用濾波條件 @temperature > 20 ，根據溫度場過濾小於或等於20的資料。
選取 [套用]。
例如，使用 temperature 來源資料中的欄位來過濾資料：
{
"operationType": "BuiltInTransformation",
"builtInTransformationSettings": {
"filter": [
{
"inputs": [
"$source.temperature ? $last" // ---------------- $1
],
"expression": "$1 > 20"
}
]
}
}
例如，使用 temperature 來源資料中的欄位來過濾資料：
builtInTransformationSettings: {
filter: [
{
inputs: [
'temperature ? $last'
]
expression: '$1 > 20'
}
]
}
如果 temperature 欄位大於 20，則會將資料傳遞至下一個階段。 如果 temperature 欄位小於或等於 20，則會篩選資料。
例如，使用 temperature 來源資料中的欄位來過濾資料：
builtInTransformationSettings:
filter:
- inputs:
- temperature ? $last # - $1
expression: "$1 > 20"
如果 temperature 欄位大於 20，則會將資料傳遞至下一個階段。 如果 temperature 欄位小於或等於 20，則會篩選資料。
對應：將資料從一個欄位移到另一個欄位
若要將資料映射到其他欄位並可選擇性轉換，請使用map操作。 將轉換定義為使用來源資料欄位的公式。
在操作經驗中，你目前可以使用計算、重命名和新增屬性轉換來映射資料。
Compute
使用 Compute 轉換將公式套用到來源資料。 此操作將公式套用於來源資料，並將結果儲存在欄位中。
在 [轉換 (選用)] 下，選取 [計算]> [新增]。
輸入必要的設定。
| 設定 |
描述 |
| 選取公式 |
從下拉式清單中選擇現有的公式，或選取 [自訂] 以手動輸入公式。 |
| 輸出 |
指定結果的輸出顯示名稱。 |
| 公式 |
輸入要套用至來源資料的公式。 |
| 描述 |
提供轉換的描述。 |
| 上一個已知的值 |
或者，如果目前值無法使用，請使用最後一個已知值。 |
在 公式 欄位輸入或編輯公式。 公式可以使用來源資料中的欄位。 輸入 @ 或選取 [Ctrl + 空格鍵] 以從下拉式清單中選擇資料點。 針對內建公式，選取 <dataflow> 預留位置以查看可用的資料點清單。
透過格式 @$metadata.user_properties.<property> 或 @$metadata.topic輸入 MQTT 元資料屬性。 請使用 @$metadata.<header>格式 輸入$metadata標頭。 只有屬於訊息標頭一部分的 MQTT 屬性才需要 $metadata 語法。 如需詳細資訊，請參閱欄位參考。
公式可以使用來源資料中的欄位。 例如，您可使用來源資料中的 temperature 欄位，將溫度轉換為攝氏度並將其儲存在 temperatureCelsius 輸出欄位中。
選取 [套用]。
重新命名
使用 Rename 轉換來重新命名資料點。 此操作將來源資料中的資料點重新命名為新名稱。 在資料流的後續階段使用新名稱。
在 [轉換 (選用)] 下，選取 [重新命名]>[新增]。
輸入必要的設定。
| 設定 |
描述 |
| 數據點 |
從下拉式清單中選取資料點，或輸入 $metadata 標頭。 |
| 新增資料點名稱 |
輸入資料點的新名稱。 |
| 描述 |
提供轉換的描述。 |
透過格式 @$metadata.user_properties.<property> 或 @$metadata.topic輸入 MQTT 元資料屬性。 請使用 @$metadata.<header>格式 輸入$metadata標頭。 只有屬於訊息標頭一部分的 MQTT 屬性才需要 $metadata 語法。 如需詳細資訊，請參閱欄位參考。
選取 [套用]。
新屬性
使用 New property transform 將新屬性加入來源資料。 此操作會為來源資料新增一個屬性。 在 data flow 的後續階段使用這個新屬性。
在 [轉換 (選用)] 下，選取 [新增屬性]> [新增]。
輸入必要的設定。
| 設定 |
描述 |
| 屬性索引鍵 |
輸入新屬性的索引鍵。 |
| 屬性值 |
輸入新屬性的值。 |
| 描述 |
提供新增屬性的描述。 |
選取 [套用]。
例如，你可以利用 temperature 來源資料中的欄位將溫度轉換成攝氏度，並儲存在 temperatureCelsius 欄位中。 使用情境化資料集中的 location 欄位來豐富來源資料。
{
"operationType": "BuiltInTransformation",
"builtInTransformationSettings": {
"map": [
{
"inputs": [
"$source.temperature ? $last" // ---------------- $1
],
"output": "temperatureCelsius",
"expression": "($1 - 32) * 5/9"
},
{
"inputs": [
"$context(assetDataset).location" // - $2
],
"output": "location"
}
]
}
}
你可以使用格式 $metadata.user_properties.<property> 或 $metadata.topic 來access MQTT 的元資料屬性。 你也可以使用 $metadata.<header>格式 輸入$metadata標頭。 如需詳細資訊，請參閱欄位參考。
例如，你可以利用 temperature 來源資料中的欄位將溫度轉換成攝氏度，並儲存在 temperatureCelsius 欄位中。 使用情境化資料集中的 location 欄位來豐富來源資料。
builtInTransformationSettings: {
map: [
{
inputs: [
'temperature'
]
output: 'temperatureCelsius'
expression: '($1 - 32) * 5/9'
}
{
inputs: [
'$context(assetDataset).location'
]
output: 'location'
}
]
}
你可以使用格式 $metadata.user_properties.<property> 或 $metadata.topic 來access MQTT 的元資料屬性。 你也可以使用 $metadata.<header>格式 輸入$metadata標頭。 如需詳細資訊，請參閱欄位參考。
例如，你可以利用 temperature 來源資料中的欄位將溫度轉換成攝氏度，並儲存在 temperatureCelsius 欄位中。 使用情境化資料集中的 location 欄位來豐富來源資料。
builtInTransformationSettings:
map:
- inputs:
- temperature # - $1
expression: "($1 - 32) * 5/9"
output: temperatureCelsius
- inputs:
- $context(assetDataset).location
output: location
若要深入了解，請參閱使用資料流程來對應資料及使用資料流程來轉換資料。
移除
預設情況下，輸出結構包含所有資料點。 使用移除轉換（ Remove transform）移除目標中的任何資料點。
在 轉換 (可選) 下，選取 移除。
選取要從輸出架構中移除的數據點。
選取 [套用]。
若要從輸出結構中移除資料點，請在data flow設定中使用 builtInTransformationSettings 屬性。 使用屬性 map 指定要移除的資料點。
{
"operationType": "BuiltInTransformation",
"builtInTransformationSettings": {
"map": [
{
"inputs": [
"*"
],
"output": "*"
},
{
"inputs": [
"weight"
],
"output": ""
}
{
"inputs": [
"weight.SourceTimestamp"
],
"output": ""
},
{
"inputs": [
"weight.Value"
],
"output": ""
},
{
"inputs": [
"weight.StatusCode"
],
"output": ""
},
{
"inputs": [
"weight.StatusCode.Code"
],
"output": ""
},
{
"inputs": [
"weight.StatusCode.Symbol"
],
"output": ""
}
]
}
}
builtInTransformationSettings: {
map: [
{
inputs: [
'*'
]
output: '*'
}
{
inputs: [
'weight'
]
output: ''
}
{
inputs: [
'weight.SourceTimestamp'
]
output: ''
}
{
inputs: [
'weight.Value'
]
output: ''
}
{
inputs: [
'weight.StatusCode'
]
output: ''
}
{
inputs: [
'weight.StatusCode.Code'
]
output: ''
}
{
inputs: [
'weight.StatusCode.Symbol'
]
output: ''
}
]
}
builtInTransformationSettings:
map:
- type: PassThrough
inputs:
- "*"
output: "*"
- inputs:
- weight
output: ""
- inputs:
- weight.SourceTimestamp
output: ""
- inputs:
- weight.Value
output: ""
- inputs:
- weight.StatusCode
output: ""
- inputs:
- weight.StatusCode.Code
output: ""
- inputs:
- weight.StatusCode.Symbol
output: ""
若要深入了解，請參閱使用資料流程來對應資料及使用資料流程來轉換資料。
根據結構描述來序列化資料
如果你想在傳送到目的地前先序列化資料，請指定結構和序列化格式。 否則，系統會將資料序列化為 JSON 並推斷出的類型。 像 Microsoft Fabric 或 Azure Data Lake 這類 Storage 端點需要架構以確保資料一致性。 支援的序列化格式為 Parquet 和 Delta。
在 Operations 體驗中，請在 data flow 端點細節中指定結構與序列化格式。 支援序列化格式的端點包括 Microsoft Fabric OneLake、Azure Data Lake Storage Gen 2、Azure Data Explorer 以及 local storage。 例如，若要將資料序列化為 Delta 格式，需將結構上傳至結構登錄檔，並在 data flow 目的端點設定中引用。
當你上傳綱要到綱要註冊表後，在資料流配置中引用它。
{
"builtInTransformationSettings": {
"serializationFormat": "Delta",
"schemaRef": "aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA>:<VERSION>"
}
}
當你將綱要上傳至綱要登錄庫後，在資料流設定中引用該綱要。
builtInTransformationSettings: {
serializationFormat: 'Delta'
schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA>:<VERSION>'
}
在你將schema上傳到架構登錄後，在資料流程配置中引用它。
builtInTransformationSettings:
serializationFormat: Delta
schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA>:<VERSION>'
如需結構描述登錄的詳細資訊，請參閱了解訊息結構描述。
目的地
若要設定 data flow 的目的地，請指定端點參考與資料目的地。 您可指定端點的資料目的地清單。
若要將資料傳送至本地 MQTT 代理以外的目的地，請建立一個 data flow 端點。 想了解方法，請參見 配置資料流端點。 如果目的地不是當地的 MQTT 代理，那它一定是來源。 若要深入瞭解如何使用本機 MQTT 訊息代理程式端點，請參閱 數據流必須使用本機 MQTT 訊息代理程式端點。
重要事項
Storage端點需要一個 schema 來進行序列化。 若要在 Microsoft Fabric 的 OneLake、Azure Data Lake Storage、Azure Data Explorer 或 Local Storage 中使用 data flow，必須指定結構參考。
選擇 data flow 端點作為目的地。
Storage端點需要一個 schema 來進行序列化。 如果你選擇 Microsoft Fabric OneLake、Azure Data Lake Storage、Azure Data Explorer 或 Local Storage 目的端點，必須指定結構參考。 例如，若要將資料以 Delta 格式序列化至 Microsoft Fabric 端點，您需要將架構上傳到架構註冊表，並在資料流目的端點設定中引用。
選取 [繼續] 以設定目的地。
輸入目的地的必要設定，包括要將資料傳送到其中的主題或資料表。 如需詳細資訊，請參閱設定資料目的地 (主題、容器或資料表)。
{
"destinationSettings": {
"endpointRef": "<CUSTOM_ENDPOINT_NAME>",
"dataDestination": "<TOPIC_OR_TABLE>" // See section on configuring data destination
}
}
destinationSettings: {
endpointRef: '<CUSTOM_ENDPOINT_NAME>'
dataDestination: '<TOPIC_OR_TABLE>' // See section on configuring data destination
}
destinationSettings:
endpointRef: <CUSTOM_ENDPOINT_NAME>
dataDestination: <TOPIC_OR_TABLE> # See section on configuring data destination
像資料來源一樣，利用資料目的地讓 data flow 端點能跨多個資料流重複使用。 資料目的地代表 data flow 端點設定中的子目錄。 例如，如果 data flow 端點是 storage 端點，那麼資料目的地就是 storage account 中的表格。 如果 data flow 端點是 Kafka 端點，則資料目的地就是 Kafka 主題。
| 端點類型 |
資料目的地意義 |
描述 |
| MQTT (或事件方格) |
主題 |
傳送資料所在的 MQTT 主題。 支援使用變數如${inputTopic} 和 ${inputTopic.index}進行靜態主題和動態主題的翻譯。 如需詳細資訊，請參閱 動態目的地主題。 |
| Kafka (或事件中樞) |
主題 |
傳送資料的 Kafka 主題。 僅支援靜態主題，不支援萬用字元。 如果端點是事件中樞命名空間，則資料目的地是該命名空間內的個別事件中樞。 |
| Azure Data Lake Storage |
容器 |
在儲存帳戶中的容器。 不是資料表。 |
| Microsoft Fabric OneLake |
資料表或資料夾 |
對應於端點所設定的路徑類型。 |
| Azure Data Explorer |
表 |
Azure Data Explorer 資料庫中的表格。 |
| 本機儲存 |
資料夾 |
本地 storage persistent volume mount 中的資料夾或目錄名稱。 當使用 由 Azure Arc 雲端 Ingest Edge Volumes 啟用的 Azure 容器儲存時，這個值必須符合您創建的子磁碟的 spec.path 參數。 |
| OpenTelemetry |
主題 |
傳送資料的 OpenTelemetry 主題。 僅支援靜態主題。 |
若要設定資料目的地：
當你使用 Operations Experience 時，它會根據端點類型自動解讀資料目的地欄位。 例如，如果 data flow 端點是 storage 端點，destination details 頁面會提示你輸入容器名稱。 如果 data flow 端點是 MQTT 端點，目的地詳細頁面會提示你輸入主題，依此類推。
{
"destinationSettings": {
"endpointRef": "<CUSTOM_ENDPOINT_NAME>",
"dataDestination": "<TOPIC_OR_TABLE>" // See section on configuring data destination
}
}
若要將資料回傳至本地的 MQTT 經紀人，請使用以下設定中的靜態 MQTT 主題：
{
"destinationSettings": {
"endpointRef": "default",
"dataDestination": "example-topic"
}
}
如果你有自訂的事件集線器端點，設定如下：
{
"destinationSettings": {
"endpointRef": "my-eh-endpoint",
"dataDestination": "individual-event-hub"
}
}
針對 MQTT 端點，您也可以使用動態主題變數。 要將訊息路由到 factory/1/dataprocessed/factory/1，請使用以下範例：
{
"destinationSettings": {
"endpointRef": "default",
"dataDestination": "processed/factory/${inputTopic.2}"
}
}
所有 data flow 端點的語法相同：
destinationSettings: {
endpointRef: "<CUSTOM_ENDPOINT_NAME>"
dataDestination: '<TOPIC_OR_TABLE>'
}
若要將資料回傳至本地的 MQTT 經紀人，請使用以下設定中的靜態 MQTT 主題：
destinationSettings: {
endpointRef: 'default'
dataDestination: 'example-topic'
}
如果你有自訂的事件集線器端點，設定如下：
destinationSettings: {
endpointRef: 'my-eh-endpoint'
dataDestination: 'individual-event-hub'
}
此範例使用一個 storage 端點作為目的地：
destinationSettings: {
endpointRef: 'my-adls-endpoint'
dataDestination: 'my-container'
}
針對 MQTT 端點，您也可以使用動態主題變數：
destinationSettings: {
endpointRef: 'default'
dataDestination: 'processed/factory/${inputTopic.2}'
}
所有 data flow 端點的語法相同：
destinationSettings:
endpointRef: <CUSTOM_ENDPOINT_NAME>
dataDestination: <TOPIC_OR_TABLE>
若要將資料回傳至本地的 MQTT 經紀人，請使用以下設定中的靜態 MQTT 主題：
destinationSettings:
endpointRef: default
dataDestination: example-topic
如果你有自訂的事件集線器端點，設定如下：
destinationSettings:
endpointRef: my-eh-endpoint
dataDestination: individual-event-hub
此範例使用一個 storage 端點作為目的地：
destinationSettings:
endpointRef: my-adls-endpoint
dataDestination: my-container
針對 MQTT 端點，您也可以使用動態主題變數：
destinationSettings:
endpointRef: default
dataDestination: processed/factory/${inputTopic.2}
動態目的地主題
對於 MQTT 端點，在欄位中使用動態主題變數 dataDestination ，根據來源主題結構路由訊息。 下列變數可供使用：
-
${inputTopic} - 完整原始輸入主題
-
${inputTopic.index} - 輸入主題的區段（索引從 1 開始）
例如， processed/factory/${inputTopic.2} 將訊息從 factory/1/data 路由傳送至 processed/factory/1。 主題段落採用 1 索引，且忽略前後斜線。
若主題變數無法解析（例如 ${inputTopic.5} 輸入主題只有三個區段），則會丟棄該訊息並記錄警告。 目的地主題中不允許通配符 （# 和 +） 。
附註
字元 $、 {和 } 在 MQTT 主題名稱中有效，因此，如果您打算使用動態主題變數，則類似 factory/$inputTopic.2 的主題是可接受的，但不正確。
範例
以下範例是一個使用 MQTT 端點作為來源與目的地的 data flow 配置。 資料來源會過濾MQTT主題azure-iot-operations/data/thermostat的資料。 轉換將溫度轉換為華氏度，並篩選溫度乘以濕度小於 100000 的資料。 目的地會將資料傳送至 MQTT 主題 factory。
使用 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": {
"dataSources": [
"thermostats/+/sensor/temperature/#",
"humidifiers/+/sensor/humidity/#"
],
"endpointRef": "default",
"serializationFormat": "Json"
}
},
{
"builtInTransformationSettings": {
"datasets": [],
"filter": [
{
"expression": "$1 * $2 < 100000",
"inputs": [
"temperature.Value",
"\"Tag 10\".Value"
],
"type": "Filter"
}
],
"map": [
{
"inputs": [
"*"
],
"output": "*",
"type": "PassThrough"
},
{
"expression": "fToC($1)",
"inputs": [
"Temperature.Value"
],
"output": "TemperatureF",
"type": "Compute"
},
{
"inputs": [
"@\"Tag 10\".Value"
],
"output": "Humidity",
"type": "Rename"
}
],
"serializationFormat": "Json"
},
"operationType": "BuiltInTransformation"
},
{
"destinationSettings": {
"dataDestination": "factory",
"endpointRef": "default"
},
"operationType": "Destination"
}
]
}
範例指令用於使用預設的資料流程配置檔來創建或更新資料流程：
az iot ops dataflow apply --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --config-file ~/data-flow.json
以下是另一個使用動態主題翻譯將訊息從不同控溫器路由傳送至裝置特定主題的範例：
{
"mode": "Enabled",
"operations": [
{
"operationType": "Source",
"sourceSettings": {
"dataSources": [
"thermostats/+/sensor/temperature"
],
"endpointRef": "default",
"serializationFormat": "Json"
}
},
{
"destinationSettings": {
"dataDestination": "processed/device/${inputTopic.2}/temperature",
"endpointRef": "default"
},
"operationType": "Destination"
}
]
}
此組態會處理來自 thermostats/device1/sensor/temperature 的訊息，並將其傳送至 processed/device/device1/temperature。
param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowName string = '<DATAFLOW_NAME>'
resource aioInstance 'Microsoft.IoTOperations/instances@2024-11-01' existing = {
name: aioInstanceName
}
resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
name: customLocationName
}
// Pointer to the default data flow endpoint
resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' existing = {
parent: aioInstance
name: 'default'
}
// Pointer to the default data flow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-11-01' existing = {
parent: aioInstance
name: 'default'
}
resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-01' = {
// Reference to the parent data flow 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'
}
}
]
}
}
apiVersion: connectivity.iotoperations.azure.com/v1
kind: Dataflow
metadata:
name: my-dataflow
namespace: azure-iot-operations
spec:
# Reference to the default data flow 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:
- 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: default
# Send the data to the MQTT topic factory
dataDestination: factory
想看更多data flow配置範例，請參考 Azure REST API - Data flow 以及 quickstart Bicep。
確認 data flow 是否正常運作
要確認資料流正常運作，請參閱 教學：建置雙向 MQTT 橋接至 Azure Event Grid。
匯出資料流設置
若要匯出資料流配置，請使用操作管理介面或匯出資料流自訂資源。
選擇你想匯出的data flow，然後從工具列選擇 Export。
操作體驗介面截圖，顯示已設定資料流的匯出選項。
使用 az iot ops dataflow show 指令來匯出資料流程。
az iot ops dataflow show --resource-group <ResourceGroupName> --instance <AioInstanceName> --name <DataflowName> --profile <DataflowProfileName> --output json > my-dataflow.json
這裡有一個範例指令，可以將一個名為 data-flow 的data flow匯出成名為 data-flow.json 的 JSON 檔案：
az iot ops dataflow show --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --output json > data-flow.json
kubectl get dataflow my-dataflow -o yaml > my-dataflow.yaml
正確的數據流（data flow）配置
為確保 data flow 如預期運作，請驗證以下條件：
- 預設的 MQTT data flow端點必須作為來源或目的節點。
-
資料流檔案 存在，並在資料流配置中被參考。
- 來源可能是 MQTT 端點、Kafka 端點或資產。 你不能用 storage 類型的端點作為來源。
- 當你使用 Event Grid 作為來源時，你會將 資料流設定檔實例數 設為 1，因為 Event Grid 的 MQTT 代理不支援共享訂閱。
- 當你使用 Event Hubs 作為來源時，命名空間中的每個事件中心都是獨立的 Kafka 主題，你必須指定每個為資料來源。
- 轉換 (如果使用的話) 已使用適當的語法進行設定，包括特殊字元的適當跳脫。
- 當您使用儲存類型端點作為目的地時，會指定一個結構。
- 當你為 MQTT 端點使用動態目的地主題時，請確保主題變數參考有效的區段。
後續步驟
