Flink Java API - Pojo 类型到元组数据类型
Flink Java API - Pojo Type to Tuple Datatype
我正在 JAVA flink API 上创建一个小实用程序来学习功能。我正在尝试读取 csv 文件并打印它,我已经为数据结构开发了一个 POJO class。当我执行代码时,我没有看到正确的值。(整数值被字符串的零和空值替换。如何映射属性的数据类型
我的主Class:
package org.karthick.flinkLab;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import javax.xml.crypto.Data;
public class CSVFileRead {
public static void main(String[] args) throws Exception {
System.out.println("--CSV File Reader using Flink's Data Set API--");
ExecutionEnvironment execEnv = ExecutionEnvironment.getExecutionEnvironment();
DataSet<DataModel> csvInput = execEnv.readCsvFile("C:\Flink\Data\IndividualDetails.csv")
.pojoType(DataModel.class);
csvInput.print();
}
}
我的 Pojo class (DataModel.class)
package org.karthick.flinkLab;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple12;
import java.io.Serializable;
import java.util.Date;
public class DataModel<T extends Tuple>
extends Tuple12<Integer,String,Date,Integer,String,String,String,String,String,String,Date,String>
implements Serializable
{
public Integer id;
public String government_id;
public Date diagnosed_date;
public Integer age;
public String detected_city;
public String detected_district;
public String detected_state;
public String nationality;
public String current_status;
public Date status_change_date;
public String notes;
public DataModel() {};
public String getNotes() {
return notes;
}
public Date getStatus_change_date() {
return status_change_date;
}
public String getCurrent_status() {
return current_status;
}
public String getNationality() {
return nationality;
}
public String getDetected_state() {
return detected_state;
}
public String getDetected_district() {
return detected_district;
}
public String getDetected_city() {
return detected_city;
}
public String gender ;
public Date getDiagnosed_date() {
return diagnosed_date;
}
public String getGender() {
return gender;
}
public Integer getAge() {
return age;
}
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public String getGovernment_id() {
return government_id;
}
public void setGovernment_id(String government_id) {
this.government_id = government_id;
}
}
当我执行 main 方法时,我没有看到正确的值。样本结果
(0,,Tue May 19 16:50:38 IST 2020,0,,,,,,,Tue May 19 16:50:38 IST 2020,)
如我所料
(2777,AP,Tue May 19 16:50:38 IST 2020,0,A,B,C,D,E,F,Tue May 19 16:50:38 IST 2020,G)
这里可能遗漏了什么?
您缺少从 CSV 到 POJO 的列映射。添加映射将起作用。列名的映射必须遵循以下两条规则:
- 列名称应与 POJO 中的名称完全相同。
- 映射中列的顺序应与 CSV 文件中的顺序完全相同。
您可以按如下方式定义映射:
DataSet<DataModel> csvInput = execEnv.readCsvFile("C:\Flink\Data\IndividualDetails.csv")
.pojoType(DataModel.class, "id", "age",.........);
它应该抛出错误,但它没有。可以是 bug
我正在 JAVA flink API 上创建一个小实用程序来学习功能。我正在尝试读取 csv 文件并打印它,我已经为数据结构开发了一个 POJO class。当我执行代码时,我没有看到正确的值。(整数值被字符串的零和空值替换。如何映射属性的数据类型
我的主Class:
package org.karthick.flinkLab;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import javax.xml.crypto.Data;
public class CSVFileRead {
public static void main(String[] args) throws Exception {
System.out.println("--CSV File Reader using Flink's Data Set API--");
ExecutionEnvironment execEnv = ExecutionEnvironment.getExecutionEnvironment();
DataSet<DataModel> csvInput = execEnv.readCsvFile("C:\Flink\Data\IndividualDetails.csv")
.pojoType(DataModel.class);
csvInput.print();
}
}
我的 Pojo class (DataModel.class)
package org.karthick.flinkLab;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple12;
import java.io.Serializable;
import java.util.Date;
public class DataModel<T extends Tuple>
extends Tuple12<Integer,String,Date,Integer,String,String,String,String,String,String,Date,String>
implements Serializable
{
public Integer id;
public String government_id;
public Date diagnosed_date;
public Integer age;
public String detected_city;
public String detected_district;
public String detected_state;
public String nationality;
public String current_status;
public Date status_change_date;
public String notes;
public DataModel() {};
public String getNotes() {
return notes;
}
public Date getStatus_change_date() {
return status_change_date;
}
public String getCurrent_status() {
return current_status;
}
public String getNationality() {
return nationality;
}
public String getDetected_state() {
return detected_state;
}
public String getDetected_district() {
return detected_district;
}
public String getDetected_city() {
return detected_city;
}
public String gender ;
public Date getDiagnosed_date() {
return diagnosed_date;
}
public String getGender() {
return gender;
}
public Integer getAge() {
return age;
}
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public String getGovernment_id() {
return government_id;
}
public void setGovernment_id(String government_id) {
this.government_id = government_id;
}
}
当我执行 main 方法时,我没有看到正确的值。样本结果
(0,,Tue May 19 16:50:38 IST 2020,0,,,,,,,Tue May 19 16:50:38 IST 2020,)
如我所料
(2777,AP,Tue May 19 16:50:38 IST 2020,0,A,B,C,D,E,F,Tue May 19 16:50:38 IST 2020,G)
这里可能遗漏了什么?
您缺少从 CSV 到 POJO 的列映射。添加映射将起作用。列名的映射必须遵循以下两条规则:
- 列名称应与 POJO 中的名称完全相同。
- 映射中列的顺序应与 CSV 文件中的顺序完全相同。
您可以按如下方式定义映射:
DataSet<DataModel> csvInput = execEnv.readCsvFile("C:\Flink\Data\IndividualDetails.csv")
.pojoType(DataModel.class, "id", "age",.........);
它应该抛出错误,但它没有。可以是 bug