如何将持续新增文件的Hadoop序列文件目录作为Apache Flink流输入源？

Absolutely! Apache Flink has solid support for ingesting continuously growing Hadoop Sequence File directories as a streaming data source. Let me break down how to implement this, based on your Flink version:

1. 推荐方案：使用Flink FileSource（Flink 1.16+） #

Starting from Flink 1.16, the unified FileSource API is the go-to choice for both batch and streaming file ingestion. It natively supports monitoring directories for new files and handles state persistence to avoid reprocessing.

Here's a step-by-step implementation example (Java):

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.sequencefile.SequenceFileFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.time.Duration;

public class SequenceFileStreamingJob {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Define the format for Hadoop Sequence Files (adjust key/value types as needed)
        SequenceFileFormat<Long, String> sequenceFileFormat = 
            SequenceFileFormat.forType(Long.class, String.class);

        // Build the FileSource that monitors the directory continuously
        FileSource<SequenceFileFormat.Record<Long, String>> source =
            FileSource.forRecordStreamFormat(sequenceFileFormat, new Path("/path/to/your/sequence-files-dir"))
                .monitorContinuously(Duration.ofSeconds(10)) // Scan for new files every 10 seconds
                .build();

        // Add the source to your streaming pipeline
        env.fromSource(source, WatermarkStrategy.noWatermarks(), "Sequence File Source")
            .map(record -> String.format("Key: %d, Value: %s", record.getKey(), record.getValue()))
            .print();

        env.execute("Continuous Sequence File Ingestion");
    }
}

Key notes for this approach:

• The monitorContinuously method configures how often Flink scans the directory for new files. Adjust the duration based on your latency requirements and resource constraints.
• Flink automatically tracks processed files via its state backend, so restarted jobs won't reprocess files they've already handled.
• Ensure your Sequence Files are finalized once written (no appends to existing files) — this ensures Flink correctly identifies them as new, complete files to process.

2. 旧版本兼容方案（Flink <1.16） #

If you're stuck on an older Flink version, you can use the legacy readFile API with continuous processing mode:

import org.apache.flink.api.java.io.SequenceFileInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode;

public class LegacySequenceFileJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SequenceFileInputFormat<Tuple2<Long, String>> inputFormat = 
            new SequenceFileInputFormat<>(Long.class, String.class);
        inputFormat.setFilePath(new Path("/path/to/your/sequence-files-dir"));

        DataStream<Tuple2<Long, String>> stream = env.readFile(
            inputFormat,
            "/path/to/your/sequence-files-dir",
            FileProcessingMode.PROCESS_CONTINUOUSLY,
            10000); // Scan interval in milliseconds

        stream.map(tuple -> String.format("Key: %d, Value: %s", tuple.f0, tuple.f1))
              .print();

        env.execute("Legacy Continuous Sequence File Ingestion");
    }
}

Important caveats for this legacy approach:

• PROCESS_CONTINUOUSLY will reprocess files if they're modified after being ingested. So make sure your files are immutable once written to avoid duplicate processing.
• You'll need to configure a reliable state backend (like RocksDB) to track processed files across job restarts.

General Best Practices #

• File Immutability: Always write complete Sequence Files to the target directory (avoid appending to existing files). This prevents unexpected reprocessing and ensures Flink can safely pick up new files.
• Scan Interval Tuning: Don't set the scan interval too short (it wastes resources) or too long (it increases processing latency). Test with your typical file arrival rate to find the sweet spot.
• State Backend: Use a persistent state backend (e.g., RocksDB) to ensure your job resumes correctly after failures without reprocessing all historical files.

内容的提问来源于stack exchange，提问作者insanely_sin