Kafka-Flink-Stream处理：是否有一种方法可以将输入文件重新加载到流处理中使用的变量中？

Question

我们计划使用Flink处理来自kafka主题的数据流（Json格式的日志）。

但是对于这种处理，我们需要使用每天都在变化的输入文件，其中的信息可以完全改变（不是格式，而是内容）。

每当这些输入文件中的一个发生变化时，我们必须将这些文件重新加载到程序中，并保持流处理继续进行。

重新加载数据的方式与现在相同：

DataSet<String> globalData = env.readTextFile("file:///path/to/file");

但是到目前为止，我还没有找到例子，也没有想出一种方法来触发流处理作业中的重新加载。

作为额外的信息，我们不会使用HDFS，而是在每个节点上使用本地文件系统，因此必须在每个节点上从本地文件重新加载。这是因为我们需要HDFS的唯一原因是这些输入文件总共只有100 mb，使用HDFS是一种过度使用。

编辑：

在阅读了更多内容后，我在几个地方发现这是一种方法：DataArtisans示例。

我试图创建一个简单的代码，从一个控制流中对一个流进行简单的更改，我得到了以下代码：

public class RichCoFlatMapExample extends EventTimeJoinHelper {

private String config_source_path = "NOT_INITIALIZED";

@Override
public void open(Configuration conf) {
    config_source_path = "first_file_path";

}

public abstract void processElement1(String one, String two, Collector<String> out) {

   config_source_path = one;
}


public abstract void processElement2(String one, String two, Collector<String> out) {
   String three = two + config_source_path;
   out.collect(three);
}

}

Answer 1

Flink可以监视一个目录，并在文件被移动到该目录时摄取它们；也许这就是你要找的。请参见文档中readfile的process_continureal选项。

但是，如果数据是在Kafka中，那么使用Flink的Kafka消费者将数据直接流到Flink中会更加自然。还有关于使用Kafka连接器的文档。Flink训练包括使用Kafka和Flink的练习。