flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Jack Huang <jackhu...@mz.com>
Subject Re: Cannot pass objects with null-valued fields to the next operator
Date Thu, 01 Sep 2016 20:53:55 GMT
Hi Stephan,

In the end I decided to specify a default value (e.g. empty string) when a
field is null.

On Mon, Aug 29, 2016 at 11:25 AM, Stephan Ewen <sewen@apache.org> wrote:

> Hi!
>
> Null is indeed not supported for some basic data types (tuples / case
> classes).
>
> Can you use Option for nullable fields?
>
> Stephan
>
>
> On Mon, Aug 29, 2016 at 8:04 PM, Jack Huang <jackhuang@mz.com> wrote:
>
>> Hi all,
>>
>> It seems like flink does not allow passing case class objects with
>> null-valued fields to the next operators. I am getting the following error
>> message:
>>
>> *Caused by: java.lang.RuntimeException: Could not forward element to next operator*
>>            at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:376)
>>            at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:358)
>>            at org.apache.flink.streaming.runtime.tasks.OperatorChain$BroadcastingOutputCollector.collect(OperatorChain.java:399)
>>            at org.apache.flink.streaming.runtime.tasks.OperatorChain$BroadcastingOutputCollector.collect(OperatorChain.java:381)
>>            at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:346)
>>            at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:329)
>>            at org.apache.flink.streaming.api.operators.StreamSource$ManualWatermarkContext.collect(StreamSource.java:340)
>>            at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecord(AbstractFetcher.java:225)
>>            at org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher.run(Kafka09Fetcher.java:239)
>>            at java.lang.Thread.run(Thread.java:745)*Caused by: java.lang.NullPointerException*
>>            at org.apache.flink.api.scala.typeutils.TraversableSerializer.copy(TraversableSerializer.scala:67)
>>            at org.apache.flink.api.scala.typeutils.TraversableSerializer.copy(TraversableSerializer.scala:32)
>>            at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:90)
>>            at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:30)
>>            at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:90)
>>            at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:30)
>>            at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:371)
>>            ... 9 more
>>
>> ‚Äč
>> This error goes away when I force all objects to not have fields with
>> null values. However, null is a valid value in my use case. Is there a way
>> to make it work? I am using flink-1.1.1.
>>
>>
>> Thanks,
>> Jack
>>
>
>

Mime
View raw message