apache-flink

1熱度

1回答

我在Java中編寫了一個Flink流作業，它加載包含訂戶數據（4列）的csv文件，然後在與訂戶數據匹配時從套接字流中讀取數據。起初我是用一個小的csv文件（8 MB）和一切工作正常： # flink run analytics-flink.jar 19001 /root/minisubs.csv /root/output.csv loaded 200000 subscribers from c

1熱度

2回答

Flink + Kafka：爲什麼我會丟失信息？

我寫了一個非常簡單的Flink流式作業，它使用FlinkKafkaConsumer082從Kafka獲取數據。 protected DataStream<String> getKafkaStream(StreamExecutionEnvironment env, String topic) { Properties result = new Properties(); resu

2熱度

1回答

flink：作業將不會以較高的taskmanager.heap.mb運行

簡單作業：kafka->flatmap->reduce->map。作業運行正常，默認值爲taskmanager.heap.mb（512Mb）。根據docs：this value should be as large as possible。由於有問題的機器有96Gb的RAM，我把它設置爲75000（任意值）。開始的工作給出了這樣的錯誤： Caused by: org.apache.flink.

3熱度

2回答

如何在flink中使用joda.time（或者如何使用typeutils.runtime.kryo）

在flink項目中，我使用了一個case class click。 case class click(date: LocalDateTime, stbId:String, channelId :Int) 此類填充數據集，它能正常工作的日期作爲一個Java 8 java.time.LocalDateTime。在java 7環境中切換到org.joda（version2.9）後，調用DataSet

3熱度

1回答

Apache Flink Streaming類型不匹配在flatMap函數中

試圖在scala 2.10.4中使用0.10.0 flink版本的流api。在試圖編譯這個第一個版本： import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment import org.apache.flink.streaming.api.scala.DataStream import org.apa

2熱度

1回答

弗林克InvalidTypesException：在的TypeVariable「類」「K」的類型不能確定

弗林克0.10.0剛剛發佈最近。我有一些代碼需要從0.9.1遷移。但出現以下錯誤： org.apache.flink.api.common.functions.InvalidTypesException：類'fi.aalto.dmg.frame.FlinkPairWorkloadOperator'中TypeVariable'K'的類型無法確定。這很可能是一種類型的刪除問題。只有在返回類型中的所有變

2熱度

2回答

flink - 羣集未使用羣集

我設置了一個3節點的羣集，可以非常均勻地分配任務（步驟？作業？），直到最近一次分配給一臺機器。拓撲（我們仍然使用這個術語以弗林克）： kafka (3 topics on different feeds) -> flatmap -> union -> map 有什麼關於這一步，將告訴集羣管理器把一切都在一臺機器？此外 - 圖像中的「未設置」值是什麼？我錯過了一些步驟？或者一些即將實施的UI功能

6熱度

1回答

Apache Flink如何實現迭代？

DAG（有向無環圖）執行大數據很常見。我想知道Apache Flink如何實現迭代。因爲它可以是一個循環圖。

1熱度

1回答

Flink - 構建作業以最大化吞吐量

我有4種類型的kafka主題和每種類型的65個主題。目標是對數據做一些簡單的窗口聚合並將其寫入數據庫。的拓撲結構將類似於：卡夫卡 - >窗口 - >減少 - > DB寫入某處在這個組合我想/需要做工會 - 或者可能的幾個（取決於每次組合多少個主題）。主題中的數據流量範圍從10K到> 200K messages/min。我有一個四節點的30個核心/節點的flink羣集。我如何構建這些拓撲來

5熱度

1回答

Apache Flink中的並行度

我可以在我們的Flink程序中的任務的不同部分設置不同程度的並行度嗎？例如，Flink如何解釋以下示例代碼？兩個自定義從業者MyPartitioner1，MyPartitioner2，將輸入數據分爲兩個4和2分區。 partitionedData1 = inputData1 .partitionCustom(new MyPartitioner1(), 1); env.setParal