flink-cep

0熱度

1回答

我從虛擬機發送數據流，以卡夫卡的測試題目用下面 public class WriteToKafka { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnviro

0熱度

1回答

不能在自定義DataSink（弗林克CEP）

初始化場我與Apache弗林克流API的一個問題。我可以設法建立整個CEP-環境與自定義數據源和源像使用標準的水槽時，「打印（）」，一切工作正常。這是我的片貌似現在： @RequiredArgsConstructor public class EventDataConsumer extends RichSinkFunction<EventData>{ private final tran

1熱度

1回答

使用Flink窗口和摺疊功能，元素缺失？

當我嘗試使用窗口和摺疊功能聚合元素時，元素的某些從獲取聚合中錯過。使用來自卡夫卡(value:0, value:1, value:2, value:3)的元素，並將它們合併爲奇數和偶數值。輸出爲：10-13之間 {even=[0, 2, 4], odd=[1, 3]} {even=[6, 8], odd=[5, 7, 9]} {even=[14, 16, 18], odd=[15, 1

0熱度

1回答

處理在弗林克CEP多個圖案並聯

我有以下情況有其發送流卡夫卡正在被由CEP引擎接收其中警告時產生的特定條件2的虛擬機對個人Stream滿意。目前，CEP是檢查兩個流上相同條件下（當心髒率> 65和呼吸率> 68）患者和如下圖所示 // detecting pattern Pattern<joinEvent, ? > pattern = Pattern.<joinEvent>begin("start")

0熱度

2回答

無法執行是由ClassNotFoundException的

引起弗林克儀表盤版本1.3.2 CEP模式我已經寫了這樣 Pattern<JoinedEvent, ?> pattern = Pattern.<JoinedEvent>begin("start") .where(new SimpleCondition<JoinedEvent>() { @Override public boolean filter(JoinedEv

0熱度

1回答

JunitTest的數據流中與弗林克斯佩克特

我創建一個測試，看看我的弗林克模式功能超時正確。我使用這個弗林克斯佩克特和我有以下測試用例： @Test public void SameDoor_TwoStatuses_OneSecondTimeoutPattern() { // Arrange long now = new Date().getTime(); DoorEvent event1 = new Doo

0熱度

1回答

將自定義源的數據以連續的方式寫入flink

這是我第一次使用Apache Flink（1.3.1）並且有問題。更詳細地說，我正在使用flink-core，flink-cep和flink-streaming庫。我的應用程序是一個Akka ActorSystem，它消耗來自RabbitMQ的消息，並且各個角色處理這些消息。在一些演員中，我想實例化來自Flink的StreamExecutionEnvironment並處理傳入的消息。因此我寫了一個

0熱度

2回答

爲什麼可以將PatternStream的相同事件發送到PatternSelectFunction和PatternTimeoutFunction？

我必須收集在3個卡夫卡採購流3個事件具有在給定的時間相同的correlationID，並能夠收集這些事件的全部或部分，如果他們遲到。我用在3的數據流中和CEP圖案聯合。但是我注意到與模式匹配的事件因此在select函數中收集的事件也會在超時函數中發送到超時函數。我不知道我做錯了什麼在我的例子，或者什麼，我聽不懂，但我期待的是那是正匹配的事件是不是也處於超時。我得到的印象是不相交的時間快照

0熱度

1回答

warnings.print（）打印以相反的順序事件（最後的情況下第一）節選第一事件中的Apache弗林克CEP

我想篩選是>在弗林克10使用以下圖案中的所有臨時事件， Pattern<MonitoringEvent, ?> warningPattern = Pattern.<MonitoringEvent>begin("first") .subtype(TemperatureEvent.class) .where(new FilterFunction<TemperatureE

0熱度

1回答

發送數據流/遙控插座弗林克程序在主機操作系統上運行

正如弗林克文檔中提到，我可以打開使用 [email protected]:~$ nc -l 12345 本地套接字，然後接收讀取來自文本服務器文本輸入它使用 DataStream<String> text = env.socketTextStream("localhost", 12345); text.print(); env.execute(); 然而，當我模擬了一些場景，所以我