如何用Flink Connectors讀寫txt文件

本篇內容主要講解“如何用Flink Connectors讀寫txt文件”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“如何用Flink Connectors讀寫txt文件”吧!

通過使用Flink DataSet Connectors 數據流連接器打開txt文件，并提供數據流輸入與輸出操作；

示例環境

java.version: 1.8.xflink.version: 1.11.1

示例數據源 （項目碼云下載）

Flink 系例 之 搭建開發環境與數據

示例模塊 （pom.xml）

Flink 系例 之 DataStream Connectors 與 示例模塊

數據流輸入

TextSource.java

package com.flink.examples.file;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.util.Collector;
import scala.Tuple7;

import java.util.Date;

/**
 * @Description 從txt文件中讀取內容輸出到DataSet中
 */
public class TextSource {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        String filePath = "D:\\Workspaces\\idea_2\\flink-examples\\connectors\\src\\main\\resources\\user.txt";
        DataSet<Tuple7<Integer, String, Integer, Integer, String, Date, Long>> dataSet = env.readTextFile(filePath)
                .flatMap(new FlatMapFunction<String, Tuple7<Integer, String, Integer, Integer, String, Date, Long>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple7<Integer, String, Integer, Integer, String, Date, Long>> out) throws Exception {
                        if (StringUtils.isNotBlank(value)) {
                            String[] values = value.split(",");
                            out.collect(Tuple7.apply(
                                    Integer.parseInt(values[0]),
                                    values[1],
                                    Integer.parseInt(values[2]),
                                    Integer.parseInt(values[3]),
                                    values[4],
                                    DateUtils.parseDate(values[5], "yyyy-MM-dd HH:mm:ss"),
                                    Long.parseLong(values[6])));
                        }
                    }
                });
        dataSet.print();
    }

}

數據流輸出

TextSink.java

package com.flink.examples.file;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.types.Row;

/**
 * @Description 將DataSet數據寫入到txt文件中
 */
public class TextSink {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //需先建立文件
        String filePath = "D:\\Workspaces\\idea_2\\flink-examples\\connectors\\src\\main\\resources\\user.txt";
        //將實體轉換為Row對象，new Row(字段個數)
        Row row = Row.of(15, "chen1", 40, 1, "CN", "2020-09-08 00:00:00", 1599494400000L);
        //轉換為dataSet
        DataSet<Row> dataSet = env.fromElements(row);
        //將內容寫入到File中，如果文件已存在，將會被復蓋
        dataSet.writeAsText(filePath, FileSystem.WriteMode.OVERWRITE).setParallelism(1);
        env.execute("fline file sink");
    }

}

數據展示

到此，相信大家對“如何用Flink Connectors讀寫txt文件”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！