使用Apache Flink执行计划任务

Question

我在parallelism 5上有一份flink的工作（目前！！）。其中一个richFlatMap流在打开（配置参数）方法中打开一个文件。在flatMap操作中，没有任何打开操作，它只是读取文件来搜索某些内容。（有一个实用程序类，它的方法类似于utilityClass.searchText（“abc”））。以下是样板代码：

public class MyFlatMap extends RichFlatMapFunction<...> {

    private MyUtilityFile myFile;

    @Override
    public void open(Configuration parameters) throws Exception {
        myFile.Open("fileLocation");
    }

    @Override
    public void flatMap(...) throws Exception {
        String text = myFile.searchText('abc');
        if (text != null) // take an action
        else // another action
    }

}

python脚本每天都会在特定时间更新此文件。因此，我还应该在flatMap流中打开新创建的文件（通过python脚本）。

我只是认为这可以通过schduledExecutorService只用一个线程池来完成。

我无法打开每个flatMap调用的文件，因为它太大了。

下面是我试图编写的样板代码：

public class MyFlatMap extends RichFlatMapFunction<...> implements Runnable {
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

    private MyUtilityFile myFile;

    @Override
    public void run() {
       myFile.Open("fileLocation");     
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        scheduler.scheduleAtFixedRate(this, 1, 1, TimeUnit.HOURS);

        myFile.Open("fileLocation");
    }

    @Override
    public void flatMap(...) throws Exception {
        String text = myFile.searchText('abc');
        if (text != null) // take an action
        else // another action
    }

}

此样板是否适用于Flink环境？如果不是，我如何以预定的方式打开文件？（没有“使用kafka更新文件发送事件并通过flink读取事件”等选项）

Answer 1

也许您可以直接实现ProcessingTimeCallback接口，它支持计时器操作

public class MyFlatMap extends RichFlatMapFunction<...> implements ProcessingTimeCallback { 
    private MyUtilityFile myFile;

 
    @Override
    public void open(Configuration parameters) throws Exception {
        scheduler.scheduleAtFixedRate(this, 1, 1, TimeUnit.HOURS);

        final long now = getProcessingTimeService().getCurrentProcessingTime();
        getProcessingTimeService().registerTimer(now + 3600000, this);

        myFile.Open("fileLocation");
    }

    @Override
    public void flatMap(...) throws Exception {
        String text = myFile.searchText('abc');
        if (text != null) // take an action
        else // another action
    }

    @Override
    public void onProcessingTime(long timestamp) throws Exception {
        myFile.Open("fileLocation");

        final long now = getProcessingTimeService().getCurrentProcessingTime();
        getProcessingTimeService().registerTimer(now + 3600000, this);
    }
}