Flink持久化List<Map<String, Object>>类型的数据到ES的object数组或nested对象数组中报错
Flink Java实体类中geopos的数据结构:
public class Index {
private List
Flink Java持久化代码:
IndexRequest rawIndexRequest = Requests.indexRequest()
.id(index.getId())
.index(indexName)
.type(elastics.getType())
.source(bean2map(index));
public static Mapbean2map(Object bean) { Map map = new HashMap<>(); try { BeanInfo b; b = Introspector.getBeanInfo(bean.getClass(),Object.class); PropertyDescriptor[] pds = b.getPropertyDescriptors(); for (PropertyDescriptor pd : pds) { String propertyName = pd.getName(); Method m = pd.getReadMethod(); Object properValue = m.invoke(bean); map.put(propertyName, properValue); } } catch (Exception e) { e.printStackTrace(); } return map; }
ES mapping中字段geopos的结构:
"geopos" : {
"type" : "object",
"properties" : {
"addr" : {
"type" : "keyword"
},
"lng" : {
"type" : "keyword"
},
"lat" : {
"type" : "keyword"
}
}
}
注意:正常情况下按照以上结构处理是没有问题的,出现问题的原因在于修改了已创建索引的mapping,具体看下面的报错和原因
报错:
Caused by: ElasticsearchException[Elasticsearch exception [type=illegal_state_exception, reason=Can't get text on a START_OBJECT at 1:1481]] at org.elasticsearch.ElasticsearchException.innerFromXContent(ElasticsearchException.java:510) at org.elasticsearch.ElasticsearchException.fromXContent(ElasticsearchException.java:421) at org.elasticsearch.ElasticsearchException.innerFromXContent(ElasticsearchException.java:451) ... 24 more
原因:
ES mapping中原来geopos字段的结构是(mapping1): geopos" : { "type" : "keyword", }
修改之后变成了(mapping2):
"geopos" : {
"type" : "object",
"properties" : {
"addr" : {
"type" : "keyword"
},
"lng" : {
"type" : "keyword"
},
"lat" : {
"type" : "keyword"
}
}
}
因为在修改之前已经根据mapping1创建了索引,修改后ES中geopos的结构变成了mapping2,但是mapping的修改针对已创建的索引是不生效的,
所以List>类型的数据进入ES的时候会报错Can't get text on a START_OBJECT