克朗斯 為世界各地的啤酒廠、飲料裝瓶商和食品生產商提供單獨的機器和完整的生產線。每天，數以百萬計的玻璃瓶、罐頭和 PET 容器在克朗斯生產線上運作。生產線是複雜的系統，可能存在許多錯誤，這些錯誤可能會導致生產線停頓並降低產量。克朗斯希望儘早（有時甚至在故障發生之前）偵測到故障，並通知生產線操作員以提高可靠性和產量。那麼如何檢測故障呢？克朗斯為其生產線配備了用於數據收集的傳感器，然後可以根據規則進行評估。克朗斯作為生產線製造商以及生產線操作員可以為機器建立監控規則。因此，飲料裝瓶商和其他操作員可以定義自己的生產線誤差範圍。過去，克朗斯使用基於時間序列資料庫的系統。主要挑戰是該系統難以調試，而且查詢代表機器的當前狀態，而不是狀態轉換。
這篇文章展示了克朗斯如何建立串流媒體解決方案來監控其生產線，基於 亞馬遜Kinesis 和 適用於 Apache Flink 的 Amazon 託管服務。這些完全託管的服務降低了使用 Apache Flink 建置串流應用程式的複雜性。 Apache Flink 託管服務管理底層 Apache Flink 元件，提供持久的應用程式狀態、指標、日誌等，Kinesis 可讓您經濟高效地處理任何規模的流資料。如果您想開始使用自己的 Apache Flink 應用程序，請查看 GitHub存儲庫 適用於使用 Flink 的 Java、Python 或 SQL API 的範例。
解決方案概述
克朗斯的生產線監控是 克朗斯車間指南 系統。它為公司所有活動的組織、優先排序、管理和記錄提供支援。如果機器停止或需要材料，他們可以通知操作員，無論操作員位於生產線的哪個位置。經過驗證的狀態監控規則已經內置，但也可以透過使用者介面由使用者定義。例如，如果監控的某個資料點違反閾值，則線路上可能會出現一條簡訊或觸發維護訂單。
狀態監控和規則評估系統基於 AWS 構建，使用 AWS 分析服務。下圖說明了該架構。
幾乎每個資料流應用程式都由五層組成：資料來源、流攝取、流儲存、流處理以及一個或多個目的地。在以下部分中，我們將深入探討每一層以及克朗斯所建構的生產線監控解決方案的工作原理。
資料來源
資料由邊緣設備上運行的服務收集，該服務讀取多種協定（例如西門子 S7 或 OPC/UA）。原始資料經過預處理以建立統一的 JSON 結構，這使得稍後在規則引擎中處理更容易。轉換為 JSON 的範例負載可能如下所示：
{
"version": 1,
"timestamp": 1234,
"equipmentId": "84068f2f-3f39-4b9c-a995-d2a84d878689",
"tag": "water_temperature",
"value": 13.45,
"quality": "Ok",
"meta": {
"sequenceNumber": 123,
"flags": ["Fst", "Lst", "Wmk", "Syn", "Ats"],
"createdAt": 12345690,
"sourceId": "filling_machine"
}
}流攝取
AWS IoT Greengrass 是一種開源物聯網 (IoT) 邊緣運行時和雲端服務。這允許您在本地處理資料並聚合和過濾設備資料。 AWS IoT Greengrass 提供可部署到邊緣的預先建置元件。生產線解決方案使用流管理器元件，該元件可以處理資料並將其傳輸到AWS目的地，例如 AWS 物聯網分析, 亞馬遜簡單存儲服務 （亞馬遜 S3）和 Kinesis。流管理器緩衝並聚合記錄，然後將其傳送到 Kinesis 資料流。
串流儲存
流儲存的作用是以容錯方式緩衝訊息，並使其可供一個或多個消費者應用程式使用。為了在 AWS 上實現這一目標，最常見的技術是 Kinesis 和 適用於Apache Kafka的Amazon託管流 （亞馬遜 MSK）。為了儲存來自生產線的感測器數據，克朗斯選擇了 Kinesis。 Kinesis 是一種無伺服器串流資料服務，可在任何規模下以低延遲運作。 Kinesis 資料流中的分片是唯一識別的資料記錄序列，其中流由一個或多個分片組成。每個分片具有 2 MB/s 的讀取能力和 1 MB/s 的寫入能力（最大 1,000 筆記錄/秒）。為了避免達到這些限制，數據應盡可能均勻地分佈在分片之間。發送到 Kinesis 的每筆記錄都有一個分區鍵，用於將資料分組到分片中。因此，您希望擁有大量分區鍵來均勻分配負載。在 AWS IoT Greengrass 上運行的串流管理器支援隨機分區鍵分配，這表示所有記錄最終都位於隨機分片中，並且負載均勻分佈。隨機分區鍵分配的缺點是記錄未依序儲存在 Kinesis 中。我們將在下一節中解釋如何解決這個問題，其中我們將討論水印。
水印
A 水印 是一種用於追蹤和測量資料流中事件時間進度的機制。事件時間是在來源處建立事件時的時間戳記。水印指示流處理應用程式的及時進度，因此具有較早或相等時間戳的所有事件都被視為已處理。這些資訊對於 Flink 提前事件時間並觸發相關計算（例如視窗評估）至關重要。可以配置事件時間和浮水印之間允許的滯後，以確定在考慮視窗完成並推進浮水印之前等待遲到資料的時間。
克朗斯的系統遍布全球，需要處理因連線中斷或其他網路限製而導致遲到的情況。他們首先監控遲到情況，並將預設的 Flink 遲到處理設定為他們在此指標中看到的最大值。他們遇到了邊緣設備的時間同步問題，這導致他們採用了更複雜的水印方式。他們為所有寄件者建立了全域浮水印，並使用最低值作為浮水印。所有傳入事件的時間戳記都儲存在 HashMap 中。當定期發出浮水印時，將使用此 HashMap 的最小值。為了避免因遺失資料而導致水印停滯，他們配置了一個 idleTimeOut 參數，它忽略早於某個閾值的時間戳記。這會增加延遲，但會提供很強的數據一致性。
public class BucketWatermarkGenerator implements WatermarkGenerator<DataPointEvent> {
private HashMap <String, WatermarkAndTimestamp> lastTimestamps;
private Long idleTimeOut;
private long maxOutOfOrderness;
}
流處理
從感測器收集數據並攝入 Kinesis 後，需要由規則引擎對其進行評估。此系統中的規則表示單一指標（例如溫度）或指標集合的狀態。為了解釋一項指標，需要使用多個數據點，這是一種有狀態計算。在本節中，我們將深入探討 Apache Flink 中的鍵控狀態和廣播狀態，以及如何使用它們來建立克朗斯規則引擎。
控制串流和廣播狀態模式
在 Apache Flink 中， 為 是指系統跨時間和跨操作持久儲存和管理資訊的能力，從而能夠在支援狀態計算的情況下處理流資料。
廣播狀態模式 允許將狀態指派給操作符的所有並行實例。因此，所有算子都具有相同的狀態，並且可以使用該相同的狀態來處理資料。可以使用控制流來攝取此唯讀資料。控制流是常規資料流，但通常具有低得多的資料速率。此模式可讓您動態更新所有運算子的狀態，使用戶能夠變更應用程式的狀態和行為，而無需重新部署。更準確地說，狀態的分配是透過使用控制流程來完成的。透過將新記錄新增至控制流程中，所有操作員都會收到此更新並使用新狀態來處理新訊息。
這允許克朗斯應用程式的用戶將新規則引入 Flink 應用程序，而無需重新啟動它。這可以避免停機，並在即時發生變化時提供出色的用戶體驗。規則涵蓋場景以檢測過程偏差。有時，機器資料並不像乍看那麼容易解釋。如果溫度感測器發送高值，這可能表示存在錯誤，但也可能是持續維護過程的影響。將指標放在上下文中並過濾某些值非常重要。這是透過一個稱為 分組.
指標分組
資料和指標的分組可讓您定義傳入資料的相關性並產生準確的結果。讓我們看一下下圖中的範例。
在步驟 1 中，我們定義兩個條件組。組別 1 收集機器狀態以及正在通過生產線的產品。第 2 組使用溫度和壓力感測器的值。條件組可以根據其接收的值具有不同的狀態。本例中，組1接收到機器正在運作的數據，選擇一公升瓶作為產品；這給了這個群組狀態 ACTIVE。第 2 組有溫度和壓力指標；兩個指標均高於其閾值超過 5 分鐘。這導致第 2 組處於 WARNING 狀態。這表示第 1 組報告一切正常，而第 2 組則不然。在步驟 2 中，將權重加入到群組中。在某些情況下這是必要的，因為團體可能會報告相互矛盾的資訊。在這種情況下，組 1 報告 ACTIVE 和第2組報告 WARNING，因此系統不清楚線路的狀態為何。在添加權重後，可以對各州進行排名，如步驟 3 所示。最後，選擇排名最高的州作為獲勝州，如步驟 4 所示。
在評估規則並定義最終機器狀態後，將進一步處理結果。採取的操作取決於規則配置；這可以是通知生產線操作員補充資料、進行一些維護，或只是儀表板上的視覺更新。系統的這一部分評估指標和規則並根據結果採取行動，稱為 規則引擎.
擴充規則引擎
透過讓使用者建立自己的規則，規則引擎可以擁有大量需要評估的規則，並且某些規則可能使用與其他規則相同的感測器資料。 Flink 是一個分散式系統，水平擴展非常好。若要將資料流指派給多個任務，您可以使用 keyBy() 方法。這允許您以邏輯方式對資料流進行分區，並將部分資料傳送到不同的任務管理器。這通常是透過選擇任意金鑰來完成的，這樣您就可以獲得均勻分佈的負載。在這種情況下，克朗斯增加了一個 ruleId 到數據點並將其用作密鑰。否則，所需的數據點將由另一個任務處理。鍵控資料流可以像常規變數一樣在所有規則中使用。
目的地
當規則更改其狀態時，訊息將發送到 Kinesis 流，然後透過 亞馬遜EventBridge 給消費者。一名消費者根據事件創建通知，並將其傳輸到生產線並提醒人員採取行動。為了能夠分析規則狀態更改，另一個服務將資料寫入 亞馬遜DynamoDB 表格用於快速訪問，並且 TTL 可以將長期歷史記錄卸載到 Amazon S3 以進行進一步報告。
結論
在這篇文章中，我們向您展示了克朗斯如何在 AWS 上建立即時生產線監控系統。 Apache Flink 託管服務使克朗斯團隊能夠透過專注於應用程式開發而不是基礎架構來快速入門。 Flink 的即時功能使克朗斯能夠將機器停機時間減少 10%，並將效率提高高達 5%。
如果您想建立自己的串流應用程序，請查看以下網站上的可用範例 GitHub存儲庫。如果您想使用自訂連接器擴展 Flink 應用程序，請參閱 使用 Apache Flink 更輕鬆地建立連接器：引入非同步接收器。 Async Sink 在 Apache Flink 版本 1.15.1 及更高版本中可用。
關於作者
弗洛里安·梅爾 是 AWS 的高級解決方案架構師和資料流專家。他是一名技術專家，透過使用 AWS 雲端服務解決業務挑戰，幫助歐洲客戶取得成功和創新。除了擔任解決方案架構師之外，弗洛里安還是一位充滿熱情的登山家，曾經攀登過歐洲一些最高的山脈。
埃米爾·迪特爾 是克朗斯的高級技術主管，專門從事資料工程，關鍵領域是 Apache Flink 和微服務。他的工作經常涉及關鍵任務軟體的開發和維護。在職業生活之外，他非常重視與家人共度美好時光。
西蒙·佩爾 是位於瑞士的 AWS 的解決方案架構師。他是一位務實的實幹家，熱衷於使用 AWS 雲端服務連接技術和人員。他特別關注數據流和自動化。除了工作之外，西蒙還享受家庭、戶外活動和山間健行。
