huangxiaofeng created BEAM-1905:
-----------------------------------
Summary: ParDoTranslation should translate a ParDo more simple?
Key: BEAM-1905
URL: https://issues.apache.org/jira/browse/BEAM-1905
Project: Beam
Issue Type: Improvement
Components: runner-flink
Reporter: huangxiaofeng
Assignee: Aljoscha Krettek
now, each Pardo will translate to parDoOperate + select(tag) + flatmap, even if
it do'st have sidelnput, and it just have one output.
it will make the transformations more complicated.
for (TaggedPValue output : outputs) {
final int outputTag = tagsToLabels.get(output.getTag());
TypeInformation outputTypeInfo = context.getTypeInfo((PCollection<?>)
output.getValue());
@SuppressWarnings("unchecked")
DataStream unwrapped = splitStream.select(String.valueOf(outputTag))
.flatMap(new FlatMapFunction<RawUnionValue, Object>() {
@Override
public void flatMap(RawUnionValue value, Collector<Object> out)
throws Exception {
out.collect(value.getValue());
}
}).returns(outputTypeInfo);
context.setOutputDataStream(output.getValue(), unwrapped);
}
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
