【天衍系列 01】深入理解Flink的 FileSource 组件：实现大规模数据文件处理

//创建一个FileSource数据源，并设置为批模式，读取完文件后结束
final FileSource<String> source = FileSource.forRecordStreamFormat(...)
        .build();

//创建一个FileSource数据源，并设置为流模式，每隔5分钟检查路径新文件，并读取
final FileSource<String> source = FileSource.forRecordStreamFormat(...)
        .monitorContinuously(Duration.ofMillis(5))  
        .build();

<!-- flink读取Text File文件依赖 start-->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-files</artifactId>
  <version>1.18.0</version>
</dependency>
<!-- flink读取Text File文件依赖 end-->
<!-- flink基础依赖 start -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-java</artifactId>
    <version>1.18.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_2.12</artifactId>
    <version>1.18.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>1.18.0</version>
</dependency>
<!-- flink基础依赖 end -->

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.time.Duration;
/**
 * 描述:
 * flink集成FileSource & forRecordStreamFormat使用 & 流模式
 * StreamFormat：从文件流中读取文件内容。它是最简单的格式实现，
 * 并且提供了许多拆箱即用的特性（如 Checkpoint 逻辑），
 * 但是限制了可应用的优化（例如对象重用，批处理等等）。
 *
 * @author 浅夏的猫
 * @version 1.0.0
 * @date 2024-02-07 15:30:22
 */
public class FileSourceRecordStreamingJob {
    public static void main(String[] args) throws Exception {
        // 创建 需要读取的文件路径Path
        Path path = new Path("D:\\flink\\file_source.txt");
        // 创建 读取文件的格式函数
        TextLineInputFormat textLineInputFormat = new TextLineInputFormat();
        // 创建 FileSource
        FileSource<String> fileSource = FileSource.
                forRecordStreamFormat(textLineInputFormat, path)
                //放开注释则使用流模式，每隔5分钟检查是否有新文件否则默认使用批模式
//                .monitorContinuously(Duration.ofMillis(5))
                .build();
        // 创建 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 添加 FileSource 到数据流
        env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "FileSource").print();
        // 执行任务
        env.execute("FileSourceRecordStreamingJob");
    }
}

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.impl.StreamFormatAdapter;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.time.Duration;
/**
 * 描述：flink集成FileSource & forBulkFileFormat使用 & 流模式
 * BulkFormat：从文件中一次读取一批记录。 它虽然是最 “底层” 的格式实现，但是提供了优化实现的最大灵活性。
 *
 * @author 浅夏的猫
 * @version 1.0.0
 * @date 2024-02-07 15:30:22
 */
public class FileSourceBulkStreamingJob {
    public static void main(String[] args) throws Exception {
        //创建 批量读取文件的格式函数，其实底层还是通过对单行文件读取
        BulkFormat<String, FileSourceSplit> bulkFormat = new StreamFormatAdapter<>(new TextLineInputFormat());
        // 创建 FileSource
        FileSource<String> fileSource = FileSource.
                forBulkFileFormat(bulkFormat, new Path("D:\\flink\\file_source.txt"))
                //放开注释则使用流模式,每隔5分钟检查是否有新文件，否则默认使用批模式
//                .monitorContinuously(Duration.ofMillis(5))
                .build();
        // 创建 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 添加 FileSource 到数据流
        env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "FileSource").print();
        // 执行任务
        env.execute("FileSourceBulkStreamingJob");
    }
}