共用方式為


在 Azure IoT 作業中設定數據流

重要

此頁面包含使用 Kubernetes 部署指令清單來管理 Azure IoT Operations 元件的指示,其處於預覽狀態。 這項功能隨附 數個限制,不應用於生產工作負載。

請參閱 Microsoft Azure 預覽版增補使用規定,以了解適用於 Azure 功能 (搶鮮版 (Beta)、預覽版,或尚未正式發行的版本) 的法律條款。

資料流程是透過選用轉換從來源到目的地取得資料的路徑。 您可以建立 數據流自定義資源或使用 Azure IoT Operations Studio 入口網站來設定數據流 。 數據流是由三個部分所組成: 來源轉換目的地

數據流的圖表,其中顯示從來源到轉換的數據流,然後是目的地。

若要定義來源和目的地,您必須設定資料流端點。 轉換是選擇性的,可以包含擴充數據、篩選數據,以及將數據對應至另一個字段等作業。

重要

每個數據流都必須有 Azure IoT 作業本機 MQTT 訊息代理程式預設端點作為來源或目的地

您可以使用 Azure IoT 作業中的作業體驗來建立數據流。 作業體驗提供可視化介面來設定數據流。 您也可以使用 Bicep 來使用 Bicep 範本檔案建立數據流,或使用 Kubernetes 使用 YAML 檔案建立數據流。

繼續閱讀以瞭解如何設定來源、轉換和目的地。

必要條件

您可以使用預設數據流配置檔和端點,立即部署 Azure IoT 作業實例。 不過,您可能想要設定資料流配置檔和端點來自定義數據流。

數據流配置檔

如果您不需要不同的數據流調整設定,請使用 Azure IoT 作業所提供的預設數據流配置檔 。 若要瞭解如何設定數據流配置檔,請參閱 設定數據流配置檔

數據流端點

數據流端點必須設定數據流的來源和目的地。 若要快速開始使用,您可以使用 本機 MQTT 訊息代理程式的預設數據流端點。 您也可以建立其他類型的數據流端點,例如 Kafka、事件中樞或 Azure Data Lake Storage。 若要瞭解如何設定每種數據流端點類型,請參閱 設定數據流端點

開始使用

擁有必要條件之後,您就可以開始建立數據流。

若要在作業體驗中建立數據流,請選取 [數據流>建立數據流]。 然後,您會看到頁面,您可以在其中設定數據流的來源、轉換和目的地。

使用作業體驗建立數據流的螢幕快照。

檢閱下列各節,了解如何設定資料流程的作業類型。

來源

若要設定數據流的來源,請指定端點參考和端點的數據源清單。 選擇下列其中一個選項作為數據流的來源。

如果預設端點未作為來源使用,則必須作為 目的地使用。 若要深入瞭解,請參閱 數據流必須使用本機 MQTT 訊息代理程式端點

選項 1:使用預設 MQTT 端點作為來源

  1. 在 [來源詳細數據] 底下,選取 [MQTT]。

    使用作業體驗選取 MQTT 作為來源端點的螢幕快照。

  2. 輸入 MQTT 來源的下列設定:

    設定 描述
    MQTT 主題 要訂閱傳入訊息的 MQTT 主題篩選。 請參閱 設定 MQTT 或 Kafka 主題
    訊息結構描述 要用來還原串行化傳入訊息的架構。 請參閱 指定架構來還原串行化數據
  3. 選取套用

選項 2:使用資產作為來源

您可以使用 資產 作為數據流的來源。 只有在作業體驗中,才能使用資產作為來源。

  1. 在 [來源詳細數據] 底下,選取 [資產]。

  2. 選取您想要作為來源端點的資產。

  3. 選取 [繼續]

    所選取資產的數據點清單隨即顯示。

    使用作業體驗選取資產作為來源端點的螢幕快照。

  4. 選取 [ 套用 ] 以使用資產作為來源端點。

使用資產做為來源時,會使用資產定義來推斷數據流的架構。 資產定義包含資產數據點的架構。 若要深入瞭解,請參閱 遠端管理資產設定。

設定之後,來自資產的數據會透過本機 MQTT 訊息代理程式到達數據流。 因此,當使用資產做為來源時,數據流會使用本機 MQTT 訊息代理程式預設端點作為實際來源。

選項 3:使用自定義 MQTT 或 Kafka 資料流端點作為來源

如果您已建立自定義 MQTT 或 Kafka 資料流端點(例如,搭配事件方格或事件中樞使用),您可以使用它作為數據流的來源。 請記住,記憶體類型端點,例如 Data Lake 或 Fabric OneLake,無法當做來源使用。

若要設定,請使用 Kubernetes YAML 或 Bicep。 將佔位元值取代為您的自訂端點名稱和主題。

作業體驗目前不支援使用自定義 MQTT 或 Kafka 端點作為來源。

設定資料來源 (MQTT 或 Kafka 主題)

您可以在來源中指定多個 MQTT 或 Kafka 主題,而不需要修改資料流端點組態。 這種彈性表示相同的端點可以跨多個數據流重複使用,即使主題不同也一樣。 如需詳細資訊,請參閱 重複使用數據流端點

MQTT 主題

當來源是 MQTT(包含事件方格)端點時,您可以使用 MQTT 主題篩選來訂閱傳入訊息。 主題篩選可以包含通配符來訂閱多個主題。 例如, thermostats/+/telemetry/temperature/# 訂閱控溫器的所有溫度遙測訊息。 若要設定 MQTT 主題篩選:

在作業體驗數據流來源詳細數據中,選取 [MQTT],然後使用 [MQTT 主題] 欄位來指定要訂閱傳入訊息的 MQTT 主題篩選。

注意

作業體驗中只能指定一個 MQTT 主題篩選條件。 若要使用多個 MQTT 主題篩選,請使用 Bicep 或 Kubernetes。

共用訂用帳戶

若要搭配 MQTT 來源使用共用訂用帳戶,您可以以 的形式 $shared/<GROUP_NAME>/<TOPIC_FILTER>指定共用訂用帳戶主題。

在作業體驗數據流來源詳細數據中,選取 [MQTT],並使用 [MQTT 主題] 欄位來指定共用訂用帳戶群組和主題。

如果數據流配置檔中的實例計數大於一個,則所有使用 MQTT 來源的數據流都會自動啟用共用訂用帳戶。 在此情況下,會新增前置 $shared 詞,並自動產生共用訂用帳戶組名。 例如,如果您有實例計數為 3 的數據流配置檔,且數據流會使用 MQTT 端點做為主題topic1設定的來源,topic2且它們會自動轉換成和 的共用訂$shared/<GENERATED_GROUP_NAME>/topic1$shared/<GENERATED_GROUP_NAME>/topic2用帳戶。

您可以在組態中明確建立名為 $shared/mygroup/topic 的主題。 不過,不建議明確新增 $shared 主題,因為 $shared 需要時會自動新增前置詞。 如果未設定數據流,數據流可以使用組名進行優化。 例如,未設定 , $share 而且數據流只需要操作主題名稱。

重要

當實例計數大於一個時,需要共用訂閱的數據流在使用事件方格 MQTT 訊息代理程式做為來源時很重要,因為它 不支援共用訂閱。 若要避免遺失訊息,請在使用事件方格 MQTT 訊息代理程式做為來源時,將數據流配置檔實例計數設定為一個。 也就是數據流是訂閱者,並從雲端接收訊息時。

Kafka 主題

當來源是 Kafka(包含事件中樞)端點時,請指定要訂閱的個別 Kafka 主題以取得傳入訊息。 不支援通配符,因此您必須以靜態方式指定每個主題。

注意

透過 Kafka 端點使用事件中樞時,命名空間中的每個個別事件中樞都是 Kafka 主題。 例如,如果您有具有兩個事件中樞的事件中樞命名空間, thermostatshumidifiers您可以將每個事件中樞指定為 Kafka 主題。

若要設定 Kafka 主題:

作業體驗目前不支援使用 Kafka 端點作為來源。

指定來源架構

使用 MQTT 或 Kafka 作為來源時,您可以指定 架構 ,以顯示作業體驗入口網站中的數據點清單。 請注意,目前不支援使用架構來還原串行化和驗證傳入訊息

如果來源是資產,則會自動從資產定義推斷架構。

提示

若要從範例數據檔產生架構,請使用 Schema Gen Helper

若要設定用來從來源還原串行化傳入訊息的架構:

在作業體驗數據流 來源詳細數據中,選取 [MQTT ],並使用 [ 訊息架構 ] 欄位來指定架構。 您可以使用 [ 上傳] 按鈕,先上傳架構檔案。 若要深入瞭解,請參閱 瞭解訊息架構

若要深入瞭解,請參閱 瞭解訊息架構

轉換

轉換作業是您可以在將數據傳送至目的地之前,先從來源轉換數據的位置。 轉換是選用的。 如果您不需要變更資料,請勿在資料流程組態中包含轉換作業。 不論在組態中指定的順序為何,多個轉換都會分階段鏈結在一起。 階段的順序一律為:

  1. 擴充:將其他數據新增至源數據,並提供數據集和條件以符合。
  2. 篩選:根據條件篩選數據。
  3. 對應計算重新命名或新增 屬性:使用選擇性轉換,將數據從一個字段移至另一個字段。

本節是數據流轉換的簡介。 如需詳細資訊,請參閱 使用數據流對應數據、 使用數據流轉換數據,以及 使用數據流擴充數據。

在作業體驗中,選取 [數據流>新增轉換][選擇性]。

使用作業體驗將轉換新增至數據流的螢幕快照。

擴充:新增參考資料

若要擴充數據,請先在 Azure IoT Operations 狀態存放區中新增參考數據集。 資料集用於根據條件,將額外的資料新增至來源資料。 條件會指定為來源資料中的欄位,該欄位符合資料集中的欄位。

您可以使用狀態存放區 CLI 將範例資料載入狀態存放區。 狀態存放區中的索引鍵名稱會對應至數據流組態中的數據集。

目前, 作業體驗不支援擴充 階段。

如果資料集具有 asset 欄位的記錄,類似於:

{
  "asset": "thermostat1",
  "location": "room1",
  "manufacturer": "Contoso"
}

具有欄位比對thermostat1的來源deviceId數據具有location篩選和對應階段中可用的 和 manufacturer 欄位。

如需條件語法的詳細資訊,請參閱 使用數據流 擴充數據和使用 數據流轉換數據。

篩選:根據條件篩選資料

若要根據條件篩選資料,您可使用 filter 階段。 條件會指定為來源資料中符合某個值的欄位。

  1. [轉換] 下,選取 [篩選>新增]。

    使用作業體驗來新增篩選轉換的螢幕快照。

  2. 輸入必要的設定。

    設定 描述
    篩選條件 根據源數據中的欄位篩選數據的條件。
    描述 提供篩選條件的描述。

    在篩選條件欄位中,輸入 @ 或選取 Ctrl + 空格 鍵,以從下拉式清單中選擇數據點。

    您可以使用 格式 @$metadata.user_properties.<property>@$metadata.topic輸入 MQTT 元資料屬性。 您也可以使用 格式 @$metadata.<header>輸入$metadata標頭。 $metadata只有屬於訊息標頭一部分的 MQTT 屬性才需要語法。 如需詳細資訊,請參閱 欄位參考

    條件可以使用源數據中的欄位。 例如,您可以使用篩選條件,例如 @temperature > 20 根據溫度欄位篩選小於或等於 20 的數據。

  3. 選取套用

對應:將資料從一個欄位移到另一個欄位

若要透過選用轉換將資料對應至另一個欄位,您可使用 map 作業。 轉換會指定為使用來源資料中欄位的公式。

在作業體驗中,目前支援使用 計算重新命名New 屬性 轉換進行對應。

計算

您可以使用 計算 轉換將公式套用至源數據。 此作業可用來將公式套用至源數據,並儲存結果欄位。

  1. [轉換] 下,選取 [計算>新增]。

    使用作業體驗來新增計算轉換的螢幕快照。

  2. 輸入必要的設定。

    設定 描述
    選取公式 從下拉式清單中選擇現有的公式,或選取 [ 自定義 ] 手動輸入公式。
    輸出 指定結果的輸出顯示名稱。
    公式 輸入要套用至源數據的公式。
    描述 提供轉換的描述。
    上一個已知的值 或者,如果目前值無法使用,請使用最後一個已知值。

    您可以在 [公式] 欄位中輸入或編輯公式。 公式可以使用源數據中的欄位。 輸入 @ 或選取 Ctrl + 空格 鍵,以從下拉式清單中選擇資料點。

    您可以使用 格式 @$metadata.user_properties.<property>@$metadata.topic輸入 MQTT 元資料屬性。 您也可以使用 格式 @$metadata.<header>輸入$metadata標頭。 $metadata只有屬於訊息標頭一部分的 MQTT 屬性才需要語法。 如需詳細資訊,請參閱 欄位參考

    公式可以使用源數據中的欄位。 例如,您可以使用 temperature 源數據中的欄位,將溫度轉換成攝氏,並將其儲存在 temperatureCelsius 輸出欄位中。

  3. 選取套用

重新命名

您可以使用重新命名轉換來重新命名資料點。 此作業可用來將源數據中的數據點重新命名為新的名稱。 新的名稱可以在數據流的後續階段中使用。

  1. [轉換] 下,選取 [重新命名>新增]。

    使用作業體驗重新命名數據點的螢幕快照。

  2. 輸入必要的設定。

    設定 描述
    Datapoint 從下拉式清單中選取數據點,或輸入$metadata標頭。
    新增數據點名稱 輸入 datapoint 的新名稱。
    描述 提供轉換的描述。

    輸入 @ 或選取 Ctrl + 空格 鍵,以從下拉式清單中選擇資料點。

    您可以使用 格式 @$metadata.user_properties.<property>@$metadata.topic輸入 MQTT 元資料屬性。 您也可以使用 格式 @$metadata.<header>輸入$metadata標頭。 $metadata只有屬於訊息標頭一部分的 MQTT 屬性才需要語法。 如需詳細資訊,請參閱 欄位參考

  3. 選取套用

新屬性

您可以使用 New 屬性轉換,將新的屬性新增至源數據。 這項作業是用來將新屬性加入源數據。 新的 屬性可以在數據流的後續階段中使用。

  1. [轉換] 下,選取 [新增] 屬性>[新增]。

    使用作業體驗來新增屬性的螢幕快照。

  2. 輸入必要的設定。

    設定 描述
    屬性索引鍵 輸入新屬性的索引鍵。
    屬性值 輸入新屬性的值。
    描述 提供新屬性的描述。
  3. 選取套用

若要深入瞭解,請參閱 使用數據流 對應數據和使用 數據流轉換數據。

根據架構串行化數據

如果您想要在將資料傳送至目的地之前串行化數據,您需要指定架構和串行化格式。 否則,數據會以 JSON 串行化,並使用推斷的類型進行串行化。 Microsoft Fabric 或 Azure Data Lake 之類的記憶體端點需要架構以確保數據一致性。 支援的串行化格式為 Parquet 和 Delta。

提示

若要從範例數據檔產生架構,請使用 Schema Gen Helper

針對作業體驗,您可以在數據流端點詳細數據中指定架構和串行化格式。 支援串行化格式的端點Microsoft Fabric OneLake、Azure Data Lake Storage Gen 2 和 Azure 數據總管。 例如,若要以 Delta 格式串行化數據,您必須將架構上傳至架構登錄,並在數據流目的地端點組態中參考它。

使用作業體驗來設定數據流目的地端點串行化的螢幕快照。

如需架構登錄的詳細資訊,請參閱 瞭解訊息架構

Destination

若要設定數據流的目的地,請指定端點參考和數據目的地。 您可以指定端點的資料目的地清單。

若要將數據傳送至本機 MQTT 訊息代理程式以外的目的地,請建立數據流端點。 若要瞭解如何,請參閱 設定數據流端點。 如果目的地不是本機 MQTT 訊息代理程式,則必須作為來源使用。 若要深入瞭解,請參閱 數據流必須使用本機 MQTT 訊息代理程式端點

重要

記憶體端點需要 架構以進行串行化。 若要搭配使用數據流搭配 Microsoft Fabric OneLake、Azure Data Lake Storage、Azure 數據總管或本機記憶體,您必須 指定架構參考

  1. 選取要作為目的地的數據流端點。

    使用作業體驗來選取事件中樞目的地端點的螢幕快照。

    記憶體端點需要 架構以進行串行化。 如果您選擇Microsoft Fabric OneLake、Azure Data Lake Storage、Azure 數據總管或本機記憶體目的地端點,您必須 指定架構參考。 例如,若要將數據串行化為 Delta 格式的 Microsoft Fabric 端點,您必須將架構上傳至架構登錄,並在數據流目的地端點組態中加以參考。

    使用作業體驗來選擇輸出架構和串行化格式的螢幕快照。

  2. 選取 [ 繼續 ] 以設定目的地。

  3. 輸入目的地的必要設定,包括要傳送數據的主題或數據表。 如需詳細資訊,請參閱 設定數據目的地(主題、容器或數據表 )。

設定資料目的地 (主題、容器或資料表)

與數據源類似,數據目的地是一種概念,用來讓數據流端點在多個數據流之間重複使用。 基本上,它代表數據流端點組態中的子目錄。 例如,如果數據流端點是記憶體端點,則數據目的地是記憶體帳戶中的數據表。 如果數據流端點是 Kafka 端點,則數據目的地是 Kafka 主題。

端點類型 數據目的地意義 描述
MQTT (或事件方格) 主題 數據傳送所在的 MQTT 主題。 僅支持靜態主題,沒有通配符。
Kafka (或事件中樞) 主題 傳送數據的 Kafka 主題。 僅支持靜態主題,沒有通配符。 如果端點是事件中樞命名空間,則數據目的地是命名空間內的個別事件中樞。
Azure Data Lake Storage 容器 記憶體帳戶中的容器。 不是數據表。
Microsoft Fabric OneLake 數據表或資料夾 對應至端點的已設定路徑類型。
Azure 資料總管 Table Azure 數據總管資料庫中的數據表。
本機存放區 資料夾 本機記憶體永續性磁碟區掛接中的資料夾或目錄名稱。 使用 由 Azure Arc Cloud Ingest Edge 磁碟區啟用的 Azure Container Storage 時,這必須符合 spec.path 您所建立子卷的參數。

若要設定資料目的地:

使用作業體驗時,數據目的地欄位會根據端點類型自動解譯。 例如,如果數據流端點是記憶體端點,目的地詳細數據頁面會提示您輸入容器名稱。 如果數據流端點是 MQTT 端點,目的地詳細數據頁面會提示您輸入主題等等。

此螢幕快照顯示提示使用者輸入指定端點類型的 MQTT 主題的作業體驗。

範例

下列範例是數據流組態,其使用來源和目的地的 MQTT 端點。 來源會篩選 MQTT 主題 azure-iot-operations/data/thermostat中的數據。 轉換會將溫度轉換成華氏,並篩選溫度乘以濕度小於 100000 的數據。 目的地會將資料傳送至 MQTT 主題 factory

如需組態範例,請參閱 Bicep 或 Kubernetes 索引標籤。

若要查看數據流設定的更多範例,請參閱 Azure REST API - 數據流快速入門 Bicep

確認數據流正常運作

遵循教學課程:雙向 MQTT 網橋,以 Azure 事件方格 以確認數據流是否正常運作。

匯出數據流組態

若要匯出數據流組態,您可以使用作業體驗,或匯出數據流自定義資源。

選取您要匯出的數據流,然後從工具欄中選取 [ 匯出 ]。

使用作業體驗匯出數據流的螢幕快照。

適當的數據流組態

若要確保數據流如預期般運作,請確認下列各項:

  • 預設 MQTT 資料流端點 必須做為來源或目的地
  • 數據流配置檔存在,且會在數據流組態中參考。
  • 來源為 MQTT 端點、Kafka 端點或資產。 記憶體類型端點無法當做來源使用。
  • 使用事件方格做為來源時,數據流配置檔實例計數會設定為1,因為事件方格 MQTT 訊息代理程式不支援共用訂閱。
  • 使用事件中樞做為來源時,命名空間中的每個事件中樞都是個別的 Kafka 主題,而且必須指定為數據源。
  • 使用時,會使用適當的語法來設定轉換,包括特殊字元的適當逸出。
  • 使用記憶體類型端點作為目的地時, 會指定架構。

下一步