本篇内容主要讲解“如何用Flink Connectors读写txt文件”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“如何用Flink Connectors读写txt文件”吧!
成都创新互联-专业网站定制、快速模板网站建设、高性价比旌德网站开发、企业建站全套包干低至880元,成熟完善的模板库,直接使用。一站式旌德网站制作公司更省心,省钱,快速模板网站建设找我们，业务覆盖旌德地区。费用合理售后完善，十余年实体公司更值得信赖。
通过使用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文件”有了更深的了解，不妨来实际操作一番吧！这里是创新互联网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！