0

我的代码使用 readTextFile 来读取日志文件，当我在 Flink ( /opt/flink-1.0.3/bin/flink run -m yarn-cluster -yn 2 /home/flink/flink-json-0.1.jar) 中运行 jar 时，它会成功处理里面的行并停止我的应用程序，而不是等待新行。我需要一些参数来做吗？

val env = StreamExecutionEnvironment.getExecutionEnvironment
val stream = env.readTextFile("hdfs:///test/ignicion.io")

先感谢您

4

1 回答 1

2

你正在寻找

StreamExecutionEnvironment.readFileStream(String filePath, long intervalMillis, WatchType watchType)

对于 WatchType，您有以下选项

  • ONLY_NEW_FILES，
  • REPROCESS_WITH_APPENDED，
  • PROCESS_ONLY_APPENDED；

来自的流

StreamExecutionEnvironment.readTextFile(String filePath, String charsetName)

将在读取所有文件后完成。我认为，它主要用于开发期间的本地测试。

于 2016-06-10T16:09:46.877 回答