flink-streaming

7熱度

1回答

在我們的項目中，我們有一個Flink（1.1.3）流作業，它從一個kafka隊列讀取數據，執行映射函數轉換並寫入另一個隊列。直到我們將流出的REST請求作爲流的一部分引入之後，這一切運行良好。要做到這一點，我們使用了PlayFramework WSClient（因爲它是在我們的堆棧的其他地方使用），並以這種方式創造了它的代碼： val config = new AhcWSClientConfi

0熱度

1回答

Flink streaming：如何控制執行時間

Spark流提供終止awaitTermination（）的API。有沒有類似的API可以在幾秒鐘後正常關閉flink流？

0熱度

1回答

kafka-apache flink執行log4j錯誤

我試圖運行一個簡單的Apache Flink腳本與卡夫卡指令，但我一直有執行問題。該腳本應該讀取來自kafka製作者的消息，詳細闡述它們，然後再發送回另一個主題，即處理結果。我從這裏得到這個例子： http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Simple-Flink-Kafka-Test-td4828.

0熱度

1回答

在Apache Flink中連續處理兩個數據源

我想用Apache Flink一個接一個地批處理兩個文件。對於一個具體的例子：假設我想給每一行分配一個索引，這樣來自第二個文件的行跟隨第一行。相反，這樣做的，下面的代碼行交錯的兩個文件： val env = ExecutionEnvironment.getExecutionEnvironment val text1 = env.readTextFile("/path/to/file1")

0熱度

1回答

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

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

1熱度

1回答

Apache Flink：如何更改緩衝區超時參數？

Apache Flink緩存任務的傳出，然後發送下一個任務進行處理。緩衝會影響延遲，因爲我知道即使沒有填充緩衝區，緩衝也會發送數據到下一個任務。我該如何改變緩衝超時？我找不到任何文件。是每個Flink羣集或每個TaskManager的配置？它可以根據任務/操作員進行配置嗎？因爲我知道Flink緩衝區，即使任務在同一個TaskManager上。在這種情況下，它將影響同一個TaskManager

0熱度

1回答

卡夫卡/ Flink與地圖功能的集成問題

我在Apache Flink的scala中運行一個簡單的腳本。我從Apache Kafka製作人處讀取數據。這是我的代碼。 import java.util.Properties import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment import org.apache.flink.stream

0熱度

1回答

阿帕奇弗林克複雜的分析數據流設計與挑戰

問題陳述：試圖評估的Apache弗林克建模先進的實時低延遲分佈分析使用案例摘要：的儀器I1提供複雜的分析，I2 ，I3 ...等各有產品定義P1，P2，P3;配置用戶參數（動態）U1，U2，U3 &需要流媒體市場數據M1，M2，M3 ... 儀器分析功能（A1，A2）在計算複雜性方面複雜，其中一些可能需要300-400ms但可以並行計算。從上面清楚地看到，市場數據流將比分析功能&需要消耗最

1熱度

1回答

Flink Table API＆SQL和地圖類型（Scala）

我在使用Flink的Table API和/或Flink的SQL支持（Flink 1.3.1，Scala 2.11）在streaming環境中。我開始用DataStream[Person]和Person是一個案例類，看起來像： Person(name: String, age: Int, attributes: Map[String, String]) 所有工作正常，直到我開始把attribut

1熱度

1回答

兩個Streams之間的Flink Scala連接似乎不起作用

我想加入來自Kafka生產者的兩個流（json）。如果我篩選數據，代碼將起作用。但是當我加入他們時似乎不起作用。我想打印到控制檯的聯合流，但沒有出現。這是我的代碼 import java.util.Properties import org.apache.flink.streaming.api.scala._ import org.apache.flink.streaming.connec