Flink使用java实现读取csv文件简单实例

首先我们来看官方文档中给出的几种方法&＃xff1a;

第一种&＃xff1a;
DataSet> csvInput &＃61; env.readCsvFile(“hdfs:///the/CSV/file”)
.types(Integer.class, String.class, Double.class);

第二种&＃xff1a;
DataSet> csvInput &＃61; env.readCsvFile(“hdfs:///the/CSV/file”)
.includeFields(“10010”)
.types(String.class, Double.class);
第三种&＃xff1a;
DataSet> csvInput &＃61; env.readCsvFile(“hdfs:///the/CSV/file”)
.pojoType(Person.class, “name”, “age”, “zipcode”);

其中第一种和第二种比较类似&＃xff0c;无非就是后面要使用types定义接受的数据类型&＃xff0c;对于includeFields() 方法&＃xff0c;里面有两种传参方式&＃xff0c;一种是includeFields(“10010”)&＃xff0c;另一种是includeFields(true&＃xff0c;false。。。)。表示我们对于这个文件需要第几列&＃xff0c;就写上1或者true即可&＃xff0c;但是一定要注意一定要对应上即可

这里我们演示第三种&＃xff0c;并且扩展一些参数的使用&＃xff1a;
第三种是用一个POJO类型来接受数据&＃xff0c;所以我们一定先要定义一个POJO的类&＃xff0c;这里有很坑爹的地方就是&＃xff0c;我们在定义属性之后&＃xff0c;一定要加上这三个东西&＃xff0c;
1.getter&＃xff0c;setter方法
2.construct构造器
3.还有就是一个空的构造器&＃xff0c;不带任何参数&＃xff0c;这个一定要加上&＃xff0c;否则java不识别这是一个POJO
4.把tostring方法也加上
其实我们文件中的内容如下&＃xff1a;

name,age,add
fanglei,20,nishi
liujun,29,lalala,
woshi,18,tashi


但是我们假设不需要中间那一列&＃xff0c;所以我们定义的POJO只有两个属性&＃xff01;

public class newPerson {private String name;private String address;public newPerson(){}public newPerson(String name, String address) {this.name &＃61; name;this.address &＃61; address;}public String getName() {return name;}public void setName(String name) {this.name &＃61; name;}public String getAddress() {return address;}public void setAddress(String address) {this.address &＃61; address;}&＃64;Overridepublic String toString() {return "newPerson{" &＃43;"name&＃61;&＃39;" &＃43; name &＃43; &＃39;\&＃39;&＃39; &＃43;", address&＃61;&＃39;" &＃43; address &＃43; &＃39;\&＃39;&＃39; &＃43;&＃39;}&＃39;;}


直接上例子

public class JavaDataSetDataSourceApp {public static void main(String[] args) throws Exception {ExecutionEnvironment env &＃61; ExecutionEnvironment.getExecutionEnvironment();csvfile(env);}public static void csvfile(ExecutionEnvironment env) throws Exception {String filePath &＃61; "C:\\Users\\77051\\Desktop\\hello.csv";DataSet csvInput &＃61; env.readCsvFile(filePath).fieldDelimiter(",").ignoreFirstLine().includeFields(true,false,true).pojoType(newPerson.class,"name","address");csvInput.print();}
}


fieldDelimiter(",")表示分隔符
ignoreFirstLine()表示是否忽略第一行
includeFields(true,false,true)表示不需要第二列
pojoType(newPerson.class,“name”,“address”);表示数据类型是newPerson&＃xff0c;使用后面这两个属性

结果输出

newPerson{name&＃61;&＃39;woshi&＃39;, address&＃61;&＃39;tashi&＃39;}
newPerson{name&＃61;&＃39;fanglei&＃39;, address&＃61;&＃39;nishi&＃39;}
newPerson{name&＃61;&＃39;liujun&＃39;, address&＃61;&＃39;lalala&＃39;}


但是有一个问题就是&＃xff0c;当我的POJO中如果定义多了一个属性&＃xff0c;用这种方法读取的话&＃xff0c;结果也会把所有的age读出来&＃xff0c;但是全部赋了0 结果如下&＃xff1a;
Person{name&＃61;‘fanglei’, age&＃61;0, address&＃61;‘nishi’}
Person{name&＃61;‘liujun’, age&＃61;0, address&＃61;‘lalala’}
Person{name&＃61;‘woshi’, age&＃61;0, address&＃61;‘tashi’}

所以解决方法其实很简单&＃xff0c;重写tostring方法即可&＃xff0c;不要输出age就行了&＃xff01;