apache-flink

0熱度

1回答

我想在基於歷史事件的流中計算Flink中基於窗口的平均值（或由我定義的任何其他函數），因此流必須是事件時間（不處理基於時間）： val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment env.setStreamTimeCharacteristic(TimeCharacteri

1熱度

1回答

Flink時間窗和幻燈片時間窗有何不同？

我正在調查Apache Flink如何工作並試圖瞭解Flink中的時間窗口。

1熱度

1回答

Flink + Kafka 0.10：如何使用Kafka消息時間戳創建表作爲字段？

我想提取由FlinkKafkaConsumer010生成的消息的時間戳作爲數據流中的值。我知道AssignerWithPeriodicWatermarks類，但這似乎只是通過DataStream API爲時間聚合的目的提取時間戳。我想在後面的Table中提供該卡夫卡消息時間戳，我可以在其上使用SQL。編輯：嘗試這樣： val consumer = new FlinkKafkaConsumer

0熱度

1回答

是否可以在Windows上使用pyflink？

有沒有人曾經成功地運行python和windows使用flink？我想下面的命令： .\bin\pyflink.bat examples\python\WordCount.py ，並收到以下錯誤 Starting execution of program Usage: ./bin/pyflink<2/3>.[sh/bat] <pathToScript>[ <pathToPackage1>[ <

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回答

Apache Flink的TypeInformation 1.3.2 BatchTableEnvironment的翻譯方法

我在Apache Flink 1.3.2中嘗試了幾個flink translate（）函數：一個來自BatchTableEnvironment，另一個來自StreamTableEnvironment。該方案給了我錯誤，我迷路了。下面的程序在運行時拋出異常（不知道它需要什麼TypeInformation）。它只是使用應該與結果所需相匹配的類型信息。但它不起作用，有幫助嗎？我想我提供所有的細節....

0熱度

1回答

Flink應用函數onWindowsWindow

我正在做一個Flink項目。該項目的主要思想是讀取JSON（網絡日誌）的數據流，將它們關聯起來，並生成一個新的JSON，它是不同JSON信息的組合。此時，我可以讀取JSON，生成一個KeyedStream（基於生成日誌的機器），然後生成一個5秒的窗口流。我想要執行的下一步是將apply函數應用到窗口併合並每個JSON的信息。我對如何去做有點困惑。我現在有該代碼是以下各項之一： DataStr

2熱度

1回答

Flink流窗口內存使用

我正在評估Flink專門爲流式窗口支持生成可能的警報。我關心的是內存使用情況，所以如果有人可以幫助，這將不勝感激。例如，該應用程序將在給定的翻滾窗口（比如說5分鐘）內從該流潛在地消耗大量的數據。在評估的時候，如果說有一百萬份符合條件的文檔，它們是否都會被加載到內存中？的一般流程是： producer -> kafka -> flinkkafkaconsumer -> table.window(

2熱度

2回答

從Flink 1.3.2升級到1.4.0 hadoop文件系統和路徑問題

我最近嘗試從Flink 1.3.2升級到1.4.0，我遇到了一些問題，無法再導入org.apache.hadoop.fs.{FileSystem, Path}。問題是發生在兩個地方： ParquetWriter： import org.apache.avro.Schema import org.apache.avro.generic.GenericRecord import org.apach

0熱度

1回答

apache flink可以用來加入巨大的非實時數據嗎？

Guy我應該通過一些公用密鑰（我們正在談論多個REST服務調用的多個sql表）與一些REST服務的json加入一些巨大的SQL表。事情是這個數據不是實時/無限流，也不認爲我可以通過連接列來排序REST服務的輸出。現在，愚蠢的做法是將所有數據和行匹配起來，但這意味着將所有內容都存儲在內存/卡桑德拉或Redis等存儲中。但是，我想知道flink是否可以使用流窗口的某個王加入說X元素（所以真的只是在RA