时间戳

首页 > 解决方案 > 将已处理文件从一个文件夹移动到 flink 中的另一个文件夹

java - 将已处理文件从一个文件夹移动到 flink 中的另一个文件夹

问题描述

我是 flink 的新手，在解决以下用例时面临一些挑战

用例说明：

我将在某个文件夹中每天收到一个带有时间戳的 csv 文件，例如输入。文件格式为 file_name_dd-mm-yy-hh-mm-ss.csv。

现在我的 flink 管道将逐行读取这个 csv 文件，并将其写入我的 Kafka 主题。

读取此文件的数据完成后，需要立即将其移动到另一个文件夹历史文件夹。

为什么我需要这个是因为：假设您的 Ververica 服务器突然或手动停止，并且如果您将所有已处理的文件都放在同一位置，那么在 ververica 重新启动后，flink 将重新读取它之前处理的所有文件。因此，为了防止这种情况，这些文件需要立即将已读取的文件移动到另一个位置。

我用谷歌搜索了很多，但没有找到任何东西，所以你能指导我实现这一目标。

让我知道是否需要其他任何东西。

标签： javaapache-flinkbatch-processingflink-streamingflink-batch

解决方案

开箱即用的 Flink 提供了监视新文件的目录并读取它们的工具 - 通过StreamExecutionEnvironment.getExecutionEnvironment.readFile（参见类似的堆栈溢出线程示例 - How to read new added file in a directory in Flink / Monitoring directory for new files with Flink for data streams， ETC。）

查看函数的源代码readFile，它调用createFileInput() 方法，该方法简单地实例化ContinuousFileMonitoringFunction并ContinuousFileReaderOperatorFactory配置源 -

addSource(monitoringFunction, sourceName, null, boundedness)
                        .transform("Split Reader: " + sourceName, typeInfo, factory);

ContinuousFileMonitoringFunction实际上是大部分逻辑发生的地方。

所以，如果我要实现你的要求，我会ContinuousFileMonitoringFunction用我自己的逻辑来扩展功能，将处理过的文件移动到历史文件夹中，并从这个函数构造源代码。

鉴于该run方法在checkpointLock-

synchronized (checkpointLock) {
    monitorDirAndForwardSplits(fileSystem, context);
}

我会说在检查点完成文件上移动到历史文件夹是安全的，这些文件的修改日早，在拆分收集时globalModificationTime会更新。monitorDirAndForwardSplits

也就是说，我将扩展ContinuousFileMonitoringFunction类并实现CheckpointListener接口，并将notifyCheckpointComplete已处理的文件移动到历史文件夹：

public class ArchivingContinuousFileMonitoringFunction<OUT> extends ContinuousFileMonitoringFunction<OUT> implements CheckpointListener {
  ...

   @Override
   public void notifyCheckpointComplete(long checkpointId) throws Exception {
          Map<Path, FileStatus> eligibleFiles = listEligibleForArchiveFiles(fs, new Path(path));
        // do move logic
     }

   /**
     * Returns the paths of the files already processed.
     *
     * @param fileSystem The filesystem where the monitored directory resides.
     */
    private Map<Path, FileStatus> listEligibleForArchiveFiles(FileSystem fileSystem, Path path) {

        final FileStatus[] statuses;
        try {
            statuses = fileSystem.listStatus(path);
        } catch (IOException e) {
            // we may run into an IOException if files are moved while listing their status
            // delay the check for eligible files in this case
            return Collections.emptyMap();
        }

        if (statuses == null) {
            LOG.warn("Path does not exist: {}", path);
            return Collections.emptyMap();
        } else {
            Map<Path, FileStatus> files = new HashMap<>();
            // handle the new files
            for (FileStatus status : statuses) {
                if (!status.isDir()) {
                    Path filePath = status.getPath();
                    long modificationTime = status.getModificationTime();
                    if (shouldIgnore(filePath, modificationTime)) {
                        files.put(filePath, status);
                    }
                } else if (format.getNestedFileEnumeration() && format.acceptFile(status)) {
                    files.putAll(listEligibleFiles(fileSystem, status.getPath()));
                }
            }
            return files;
        }
    }
}

然后使用自定义函数手动定义数据流：

ContinuousFileMonitoringFunction<OUT> monitoringFunction =
                new ArchivingContinuousFileMonitoringFunction <>(
                        inputFormat, monitoringMode, getParallelism(), interval);

ContinuousFileReaderOperatorFactory<OUT, TimestampedFileInputSplit> factory = new ContinuousFileReaderOperatorFactory<>(inputFormat);

final Boundedness boundedness = Boundedness.CONTINUOUS_UNBOUNDED;

env.addSource(monitoringFunction, sourceName, null, boundedness)
                        .transform("Split Reader: " + sourceName, typeInfo, factory);

推荐阅读