flink-streaming

0熱度

1回答

從下面的文章我們知道非鍵控流的並行性是1，那麼我們可以通過setParallelism增加並行性嗎？「在非鍵控流的情況下，您的原始流不會被拆分爲多個邏輯流，並且所有窗口邏輯將由一個任務執行，即並行性爲1。 https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/windows.html#keyed-vs-non-keyed-

0熱度

1回答

GlobalWindows在Flink中的相同GlobalWindows中爲相同的鍵分配元素？

基於以下文件「全局窗口分配器分配使用相同的密鑰相同的單一全局窗口中的所有元素」 https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/windows.html 然後我檢查了源代碼，發現GlobalWindows的assignWindows方法只返回全球Window並沒有爲參數元素做任何事情，那麼所有具有相同鍵的元素如何到同一

1熱度

1回答

我們可以在Flink中結合兩者並計數和處理時間觸發器嗎？

我希望在計數達到100次或每5秒翻轉處理時間後使Windows完成？也就是說，當元素達到100時，觸發Windows計算，但是如果元素沒有達到100，但時間經過5秒，它也會觸發Windows計算，就像下面兩個觸發器的組合一樣： .countWindow（100） .window（TumblingProcessingTimeWindows.of（Time.seconds（5）））

1熱度

2回答

閃爍流窗口觸發器

我有flink流，我在一些時間窗口上說幾件事說30秒。這裏發生了什麼，它給了我結果我彙總以前的窗口以及。說前30秒，我得到的結果10 下一個THIRY秒我要新鮮的結果，而不是我得到的最後一個窗口結果+新等。所以我的問題是我如何獲得每個窗口的新鮮結果。

0熱度

1回答

如何解決在使用彈性搜索連接器將流數據發送到彈性搜索索引時flink中存在太多打開文件異常？

我正在使用flink彈性搜索連接器將數據發送到彈性搜索索引。但是我無法運行這個工作並得到這個異常。如何解決這個問題？有沒有我缺少的設置或配置？誰能指導我？ io.netty.channel.ChannelException: The operating system does not offer enough file handles to open the network connection

3熱度

2回答

如何停止程序中的閃存串流作業

我正在嘗試爲Flink串流作業創建JUnit測試，該工作將數據寫入kafka主題，並分別使用FlinkKafkaProducer09和FlinkKafkaConsumer09從同一個kafka主題讀取數據。我的產品通過測試數據： DataStream<String> stream = env.fromElements("tom", "jerry", "bill"); 以及檢查相同的數據從消費者

0熱度

1回答

爲什麼通過這種方式爲Flink計算的Key的狀態位置？

我讀了如何計算陣列爲重點的狀態位置弗林克的源代碼，並發現keyGroupIndex-keyGroupOffset計算的狀態位置，我的問題是：爲什麼要使用keyGroupIndex-keyGroupOffset作爲位置，爲什麼不直接使用狀態[keyGroupIndex]？此外，如果直接使用state[keyGroupIndex]，我還發現狀態數組的大小由Number of KeyGroup指定

1熱度

1回答

flink是否支持暫停flink作業？

我剛剛開始學習apache flink並遇到以下問題：如何暫停flink作業然後恢復它？ flink支持使用命令行暫停作業嗎？

1熱度

1回答

Flink流過去1小時，1周和1月數據的流程事件

我必須用最後1小時，1周和1個月的數據處理每個事件。就像在同一事件發生的最後1個月內發生了多少次相同的IP一樣。我想窗口是固定的時間，我不能對應當前事件最近一小時計算。如果您有任何線索請指導我應該怎麼用表，ProcessFunction或全局窗口。或者我應該採取什麼方法？

2熱度

1回答

Apache Flink上的吞吐量和延遲

我已經爲Apache Flink編寫了一個非常簡單的java程序，現在我有興趣測量統計信息，如吞吐量（每秒處理的元組數量）和延遲（程序需要處理的時間輸入元組）。 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.readTextFile("/home/Lizard