如何用FlinkConnectors读写txt文件
本篇内容主要讲解“如何用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 DataSetdataSet = env.fromElements(row); //将内容写入到File中,如果文件已存在,将会被复盖 dataSet.writeAsText(filePath, FileSystem.WriteMode.OVERWRITE).setParallelism(1); env.execute("fline file sink"); } }
数据展示
到此,相信大家对“如何用Flink Connectors读写txt文件”有了更深的了解,不妨来实际操作一番吧!这里是创新互联网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!
网站名称:如何用FlinkConnectors读写txt文件
本文路径:http://scyanting.com/article/pscece.html