目錄

1、前言

2、readTextFile（已過時，不推薦使用）

3、readFile（已過時，不推薦使用）

4、fromSource(FileSource) 推薦使用

1、前言

思考: 讀取文件時可以設(shè)置哪些規(guī)則呢？

?????????1. 文件的格式(txt、csv、二進(jìn)制...)????????

? ? ? ? ?2. 文件的分隔符(按\n 分割)

? ? ? ? ?3. 是否需要監(jiān)控文件變化(一次讀取、持續(xù)讀取)

基于以上規(guī)則，F(xiàn)link為我們提供了非常靈活的 讀取文件的方法

2、readTextFile（已過時，不推薦使用）

語法說明:

定義:
    def readTextFile(filePath: String): DataStream[String]
    def readTextFile(filePath: String, charsetName: String)

功能:
    1.讀取文本格式的文件
    2.按行讀取(\n為分隔符)，每行數(shù)據(jù)被封裝為 DataStream 的一個元素
    3.可以指定字符集(默認(rèn)為UDF-8)
    4.文件只會讀取一次

源碼分析：
    public DataStreamSource<String> readTextFile(String filePath, String charsetName) {

        // 初始化 TextInputFormat對象
        TextInputFormat format = new TextInputFormat(new Path(filePath));  
        // 指定路徑過濾器(使用默認(rèn)過濾器)
        format.setFilesFilter(FilePathFilter.createDefaultFilter());  
        // 指定Flink中的數(shù)據(jù)類型    
        TypeInformation<String> typeInfo = BasicTypeInfo.STRING_TYPE_INFO; 
        // 指定字符集
        format.setCharsetName(charsetName);     
                                   
        // 調(diào)用 readFile 方法
        return readFile(format, filePath, FileProcessingMode.PROCESS_ONCE, -1, typeInfo); 
    }

代碼示例:

    public static void readTextFile() throws Exception {
        /*
         * TODO 功能說明
         *   readTextFile(path) - 讀取文本文件(一次讀取)，例如遵守 TextInputFormat 規(guī)范的文件，逐行讀取并將它們作為字符串返回。
         * */
        // 1.獲取執(zhí)行環(huán)境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2.將文本文件作為數(shù)據(jù)源
        env.readTextFile("data/1.txt").setParallelism(4).print();

        // 3.觸發(fā)程序執(zhí)行
        env.execute();
    }

3、readFile（已過時，不推薦使用）

語法說明:

定義:
    def readFile[T: TypeInformation](
        inputFormat: FileInputFormat[T],
        filePath: String,
        watchType: FileProcessingMode,
        interval: Long): DataStream[T] = {
      val typeInfo = implicitly[TypeInformation[T]] // 隱私轉(zhuǎn)換(將java 數(shù)據(jù)類型 轉(zhuǎn)換為 Flink數(shù)據(jù)類型)
      asScalaStream(javaEnv.readFile(inputFormat, filePath, watchType, interval, typeInfo))
    }

參數(shù):
    inputFormat : 指定 FileInputFormat 實現(xiàn)類(根據(jù)文件類型 選擇相適應(yīng)的實例)
    filePath    : 指定 文件路徑
    watchType   : 指定 讀取模式(提供了2個枚舉值)
                       PROCESS_ONCE ：只讀取一次
                       PROCESS_CONTINUOUSLY ：按照指定周期掃描文件
    interval    : 指定 掃描文件的周期(單位為毫秒)

功能:
    按照 指定的 文件格式 和 讀取方式 讀取數(shù)據(jù)

代碼示例:文章來源地址http://www.zghlxwxcb.cn/news/detail-635748.html

    public static void readFile() throws Exception {
        /*
         * TODO 功能說明
         *    readFile(fileInputFormat, path) - 按照指定的文件輸入格式讀取（一次）文件。
         *    readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo)
         *       按照指定的文件輸入格式讀?。ǔ掷m(xù)的讀?。┪募?         * */

        // 1.獲取執(zhí)行環(huán)境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2.將文本文件作為數(shù)據(jù)源
        String filePath = "data/1.txt";

        TextInputFormat textInputFormat = new TextInputFormat(new Path(filePath));
        textInputFormat.setFilesFilter(FilePathFilter.createDefaultFilter()); // 指定過濾器
        textInputFormat.setCharsetName("UTF-8"); // 指定編碼格式

        /*
         * readFile(inputFormat: FileInputFormat[OUT], filePath: String, watchType: FileProcessingMode, interval: Long)
         * 參數(shù)說明：
         *      @inputFormat ： 指定文件輸入格式
         *      @filePath    ： 指定文件路徑
         *      @watchType   ： 指定監(jiān)控類型，提供了兩種讀取策略
         *            PROCESS_ONCE ： 只讀取一次
         *            PROCESS_CONTINUOUSLY ：持續(xù)讀取，監(jiān)控新增數(shù)據(jù)
         *      @interval ： 指定連續(xù)掃描文件的周期(毫秒)
         * 重點提示：
         *      1.如果watchType設(shè)置為PROCESS_CONTINUOUSLY時，當(dāng)一個文件被修改時，將會導(dǎo)致重新讀取該
         *           文件的全部內(nèi)容，這將會打破`精確一次`的語義
         * */
        env.readFile(
                textInputFormat
                , filePath
                , FileProcessingMode.PROCESS_CONTINUOUSLY
                , 1000
        ).print();

        // 3.觸發(fā)程序執(zhí)行
        env.execute();
    }

4、fromSource(FileSource) 推薦使用

    public static void FileSource() throws Exception {
        // 1.獲取執(zhí)行環(huán)境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2.將文本文件作為數(shù)據(jù)源
        FileSource<String> fileSource = FileSource.forRecordStreamFormat(new TextLineInputFormat()
                , new Path("data/1.txt")).build();

        env.fromSource(fileSource
                , WatermarkStrategy.noWatermarks()
                , "read fileSource"
        ).print();

        // 3.觸發(fā)程序執(zhí)行
        env.execute();
    }

到了這里，關(guān)于4.2、Flink任務(wù)怎樣讀取文件中的數(shù)據(jù)的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！