作者:无聊人生918_644 | 来源:互联网 | 2023-09-15 18:31
Connectors如何读写csv文件,相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决这个问题。
通过使用Flink DataSet Connectors 数据流连接器打开csv文件,并提供数据流输入与输出操作;
示例环境
java.version: 1.8.xflink.version: 1.11.1
示例数据源 (项目码云下载)
Flink 系例 之 搭建开发环境与数据
示例模块 (pom.xml)
Flink 系例 之 DataStream Connectors 与 示例模块
数据流输入
CsvSource.java
package com.flink.examples.file;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple7;
import java.util.HashMap;
import java.util.Map;
/**
* @Description 从csv文件中读取内容输出到DataSet中
*/
public class CsvSource {
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.csv";
DataSet> dataSet = env
.readCsvFile(filePath)
.fieldDelimiter(",")
.types(Integer.class, String.class, Integer.class, Integer.class, String.class, String.class, Long.class);
//打印
dataSet.print();
}
}
数据流输出
CsvSink.java
package com.flink.examples.file;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.core.fs.FileSystem;
/**
* @Description 将DataSet数据写入到csv文件中
*/
public class CsvSink {
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.csv";
//添加数据
Tuple7 row = new Tuple7<>(15, "chen1", 40, 1, "CN", "2020-09-08 00:00:00", 1599494400000L);
//转换为dataSet
DataSet> dataSet = env.fromElements(row);
//将内容写入到File中,如果文件已存在,将会被复盖
dataSet.writeAsCsv(filePath, FileSystem.WriteMode.OVERWRITE).setParallelism(1);
env.execute("fline file sink");
}
}
数据展示
看完上述内容,你们掌握Connectors如何读写csv文件的方法了吗?如果还想学到更多技能或想了解更多相关内容,欢迎关注编程笔记行业资讯频道,感谢各位的阅读!