千家信息网

如何用Flink Connectors读写txt文件

发表于:2025-01-23 作者:千家信息网编辑
千家信息网最后更新 2025年01月23日,本篇内容主要讲解"如何用Flink Connectors读写txt文件",感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习"如何用Flink Connectors
千家信息网最后更新 2025年01月23日如何用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> dataSet = env.readTextFile(filePath)                .flatMap(new FlatMapFunction>() {                    @Override                    public void flatMap(String value, Collector> 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 dataSet = env.fromElements(row);        //将内容写入到File中,如果文件已存在,将会被复盖        dataSet.writeAsText(filePath, FileSystem.WriteMode.OVERWRITE).setParallelism(1);        env.execute("fline file sink");    }}

数据展示

到此,相信大家对"如何用Flink Connectors读写txt文件"有了更深的了解,不妨来实际操作一番吧!这里是网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!

0