【Flink异常】Caused by: java.lang.RuntimeException: Row arity of from does not match serializers

文章目录

1. 异常详情
2. 分析原因
3. 解决办法

1. 异常详情

Caused by: java.lang.RuntimeException: Row arity of from does not match serializers.at org.apache.flink.api.java.typeutils.runtime.RowSerializer.copy(RowSerializer.java:89)at org.apache.flink.api.java.typeutils.runtime.RowSerializer.copy(RowSerializer.java:47)at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:577)at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554)at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534)at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:718)at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:696)at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:51)

2. 分析原因

① 查看报错源码位置

Row的长度（arity）与fieldSerializers.length长度不一致导致的报错.

② debug源码，查看两个from.getArity()与len的具体值

可知Row的长度为6（根据自己代码中有new Row(6)）可知这个值是正确的，
那fieldSerializers.length =7这个值又从何而来？

③ 分析fieldSerializers长度异常的原因

根据fieldSerializers debug内容可知，其长度为7，类型为TypeSerializer[]（StringSerializer）。应为Row的TypeInfomation序列化的结果。

报错原因：
Row的长度与TypeInformation的长度不一致

3. 解决办法

Flink改变schema时（修改字段个数或字段名），不使用原生的TypeInfomation，需要自己生成TypeInformation，来指定给转变后的DataStream。

笔者在代码中使用方式如下：（注：TestWindowFunction implements WindowFuntion）

dataStream.countWindow(1000)
.apply(new TestWindowFunction(), testRowTypeInfo);

笔者在TestWindowFunction中有new Row(6)，而指定TypeInformation时testRowTypeInfo的长度为7，导致出现Row arity of from does not match serializers异常。

因此，Flink datastream 需要保证Row的长度与TypeInfomation字段个数相等。