以下错误,想必在做Spark的DateSet操作时一定是见过吧?
Error:(58, 17) Unable to find encoder for type stored in a Dataset. Primitive types (Int, String, etc) and Product types (case classes) are supported by importing spark.implicits._ Support for serializing other types will be added in future releases.
peopleDF.map(row => row.get(0)).show()
这是因为在作map转换时需要指定一个转换的Encorder,在Scala代码中是通过隐式转换进行的,而在Java代码中则需要在代码中指明。
为了更好理解写了一个Java的代码供学习加深理解。
代码如下:
public static void main(String[] args) {
SparkSession spark = SparkSession.builder().master("local[2]").appName("DataSetEncoderExample").getOrCreate();
List<String> data = Arrays.asList("chen", "li", "huang");
//创建DataSet的时候指明数据是String类型
Dataset<String> ds = spark.createDataset(data, Encoders.STRING());
/***操作一:**/
//map操作:把string类型的变换成string类型
//此时MapFunction<String, String> 这两个地方都应该是String
Dataset<String> dsString2String = ds.map((MapFunction<String, String>) v -> "Hi," + v, Encoders.STRING());
dsString2String.show();
/***操作二:**/
//map操作:把string类型的变换成int类型
//注意此时MapFunction<String, Integer> 这两个地方的类型变化
//第一个类型String为原来的DataSet的类型,第二个类型为输出的类型
Dataset<Integer> dsString2Int = ds.map(new MapFunction<String, Integer>(){
@Override
public Integer call(String value) throws Exception {
return value.length();
}
}, Encoders.INT());
dsString2Int.show();
/***操作三:**/
//map操作:把string类型的变换成自定义的对象类型
//注意此时MapFunction<String, People> 这两个地方的类型变化
//第一个类型String为原来的DataSet的类型,第二个类型People为输出的类型
Encoder<People> peopleEncoder = Encoders.kryo(People.class);
Dataset<People> dsString2Object = ds.map(new MapFunction<String, People>(){
@Override
public People call(String value) throws Exception {
return new People(value, value.length());
}
}, peopleEncoder);
dsString2Object.show();
dsString2Object.map((MapFunction<People, String>) item -> item.getName(), Encoders.STRING()).show();
/***操作四:**/
//map操作:把string类型的变换成Row对象类型
//注意此时MapFunction<String, Row> 这两个地方的类型变化
//第一个类型String为原来的DataSet的类型,第二个类型Row为输出的类型
// Encoder<Row> rowEncoder = Encoders.kryo(Row.class);
Encoder<Row> rowEncoder = Encoders.javaSerialization(Row.class);
Dataset<Row> dsString2Row = ds.map(
(MapFunction<String, Row>) value -> RowFactory.create(value, value.length())
, rowEncoder);
dsString2Row.show();
dsString2Row.map((MapFunction<Row, String>) item -> item.getString(0), Encoders.STRING()).show();
spark.stop();
}
上述代码中需要创建一下Java类
public class People {
String name;
Integer age;
public People( String name, Integer age){
this.name = name;
this.age = age;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public Integer getAge() {
return age;
}
public void setAge(Integer age) {
this.age = age;
}
}