oh .. and I am using Flink 1.8 ..
On Mon, Aug 26, 2019 at 12:09 AM Debasish Ghosh <ghosh.debas...@gmail.com>
wrote:
> Thanks for the feedback .. here are the details ..
>
> Just to give u some background the original API is a Scala API as follows
> ..
>
> final def readStream[In: TypeInformation: DeserializationSchema](inlet:
> CodecInlet[In]): DataStream[In] =
> context.readStream(inlet)
>
> and the *Scala version of the code runs fine* .. Here's the Java API
> (also written in Scala though but passing type information and
> deserialization schema explicitly and using the DataStream class from Flink
> Java) ..
>
> final def readStream[In](inlet: CodecInlet[In], clazz: Class[In],
> deserializationSchema: DeserializationSchema[In]): JDataStream[In] =
> context.readStream(inlet)(TypeInformation.of[In](clazz),
> deserializationSchema)
> .javaStream
>
> Here's the Java code for transformation where I get the error ..
>
> DataStream<Data> ins =
> this.<Data>readStream(in, Data.class, serdeData)
> .map((Data d) -> d)
> .returns(new TypeHint<Data>(){}.getTypeInfo());
>
> DataStream<Simple> simples = ins.map((Data d) -> new Simple(d.getName()));
> // .returns(new TypeHint<Simple>(){}.getTypeInfo());
> DataStreamSink<Simple> sink = writeStream(out, simples, Simple.class,
> serdeSimple);
>
> Here's the corresponding Scala code that runs fine ..
>
> val ins: DataStream[Data] = readStream(in)
> val simples: DataStream[Simple] = ins.map(r ⇒ new Simple(r.getName()))
> writeStream(out, simples)
>
> Here's the custom source that's also referred in the exception .. the case
> class is directly used in Scala while I use the Java API that uses that
> case class from Java ..
>
> object FlinkSource {
> case class CollectionSourceFunction[T](data: Seq[T]) extends
> SourceFunction[T] {
> def cancel(): Unit = {}
> def run(ctx: SourceContext[T]): Unit = {
> data.foreach(d ⇒ ctx.collect(d))
> }
> }
>
> /**
> * Java API
> */
> def collectionSourceFunction[T](data: java.util.List[T]) =
> CollectionSourceFunction(data.asScala.toSeq)
> }
>
> Here's how I use the custom source from Java .. (which gives exception) ..
> here data is a java.util.List<Data>
>
> env.<Data>addSource(
> FlinkSource.<Data>collectionSourceFunction(data)
> )
>
> and here's the Scala version, which runs fine .. here data is a
> scala.Seq[Data]
>
> env.addSource(FlinkSource.CollectionSourceFunction(data))
>
> Here's the complete exception ..
>
> [info] org.apache.flink.api.common.functions.InvalidTypesException: The
> return type of function 'Custom Source' could not be determined
> automatically, due to type erasure. You can give type information hints by
> using the returns(...) method on the result of the transformation call, or
> by letting your function implement the 'ResultTypeQueryable' interface.
> [info] at
> org.apache.flink.streaming.api.transformations.StreamTransformation.getOutputType(StreamTransformation.java:420)
> [info] at
> org.apache.flink.streaming.api.datastream.DataStream.getType(DataStream.java:175)
> [info] at
> org.apache.flink.streaming.api.datastream.DataStream.map(DataStream.java:587)
> [info] at
> pipelines.flink.FlinkStreamletLogic.readStream(FlinkStreamlet.scala:237)
> [info] at
> pipelines.flink.javadsl.FlinkProcessorJ$1.buildExecutionGraph(FlinkProcessorJ.java:38)
> [info] at
> pipelines.flink.FlinkStreamletLogic.executeStreamingQueries(FlinkStreamlet.scala:282)
> [info] at pipelines.flink.FlinkStreamlet.run(FlinkStreamlet.scala:151)
> [info] at
> pipelines.flink.testkit.FlinkTestkit.doRun(FlinkTestkit.scala:146)
> [info] at
> pipelines.flink.testkit.FlinkTestkit.run(FlinkTestkit.scala:138)
> [info] at
> pipelines.flink.javadsl.FlinkStreamletTest.shouldProcessDataWhenItIsRun(FlinkStreamletTest.java:46)
> [info] ...
> [info] Cause:
> org.apache.flink.api.common.functions.InvalidTypesException: Type of
> TypeVariable 'T' in 'class
> pipelines.flink.testkit.FlinkSource$CollectionSourceFunction' could not be
> determined. This is most likely a type erasure problem. The type extraction
> currently supports types with generic variables only in cases where all
> variables in the return type can be deduced from the input type(s).
> Otherwise the type has to be specified explicitly using type information.
> [info] at
> org.apache.flink.api.java.typeutils.TypeExtractor.createTypeInfoWithTypeHierarchy(TypeExtractor.java:882)
> [info] at
> org.apache.flink.api.java.typeutils.TypeExtractor.privateCreateTypeInfo(TypeExtractor.java:803)
> [info] at
> org.apache.flink.api.java.typeutils.TypeExtractor.createTypeInfo(TypeExtractor.java:769)
> [info] at
> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.addSource(StreamExecutionEnvironment.java:1459)
> [info] at
> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.addSource(StreamExecutionEnvironment.java:1414)
> [info] at
> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.addSource(StreamExecutionEnvironment.java:1396)
> [info] at
> pipelines.flink.javadsl.FlinkStreamletTest.shouldProcessDataWhenItIsRun(FlinkStreamletTest.java:34)
> [info] at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> [info] at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> [info] at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> [info] ...
>
> regards.
>
> On Sun, Aug 25, 2019 at 11:44 PM Rong Rong <walter...@gmail.com> wrote:
>
>> I am not sure how the function `readStream` is implemented (also which
>> version of Flink are you using?).
>> Can you share more information on your code blocks and exception logs?
>>
>> Also to answer your question, DataStream return type is determined by its
>> underlying transformation, so you cannot set it directly.
>>
>> Thanks,
>> Rong
>>
>> On Sat, Aug 24, 2019 at 12:29 PM Debasish Ghosh <ghosh.debas...@gmail.com>
>> wrote:
>>
>>> Thanks .. I tried this ..
>>>
>>> DataStream<Data> ins = readStream(in, Data.class, serdeData).map((Data
>>> d) -> d).returns(new TypeHint<Data>(){}.getTypeInfo());
>>>
>>> But still get the same error on this line ..
>>>
>>> (BTW I am not sure how to invoke returns on a DataStream and hence had
>>> to do a fake map - any suggestions here ?)
>>>
>>> regards.
>>>
>>> On Sat, Aug 24, 2019 at 10:26 PM Rong Rong <walter...@gmail.com> wrote:
>>>
>>>> Hi Debasish,
>>>>
>>>> I think the error refers to the output of your source instead of your
>>>> result of the map function. E.g.
>>>> DataStream<Data> ins = readStream(in, Data.class, serdeData)*.returns(new
>>>> TypeInformation<Data>);*
>>>> DataStream<Simple> simples = ins.map((Data d) -> new
>>>> Simple(d.getName())).returns(new TypeHint<Simple>(){}.getTypeInfo());
>>>>
>>>> --
>>>> Rong
>>>>
>>>> On Fri, Aug 23, 2019 at 9:55 AM Debasish Ghosh <
>>>> ghosh.debas...@gmail.com> wrote:
>>>>
>>>>> Hello -
>>>>>
>>>>> I have the following call to addSource where I pass a Custom
>>>>> SourceFunction ..
>>>>>
>>>>> env.<Data>addSource(
>>>>> new CollectionSourceFunctionJ<Data>(data,
>>>>> TypeInformation.<Data>of(new TypeHint<Data>(){}))
>>>>> )
>>>>>
>>>>> where data is List<Data> and CollectionSourceFunctionJ is a Scala
>>>>> case class ..
>>>>>
>>>>> case class CollectionSourceFunctionJ[T](data: java.util.List[T], ti:
>>>>> TypeInformation[T]) extends SourceFunction[T] {
>>>>> def cancel(): Unit = {}
>>>>> def run(ctx: SourceContext[T]): Unit = {
>>>>> data.asScala.foreach(d ⇒ ctx.collect(d))
>>>>> }
>>>>> }
>>>>>
>>>>> When the following transformation runs ..
>>>>>
>>>>> DataStream<Data> ins = readStream(in, Data.class, serdeData);
>>>>> DataStream<Simple> simples = ins.map((Data d) -> new
>>>>> Simple(d.getName())).returns(new TypeHint<Simple>(){}.getTypeInfo());
>>>>>
>>>>> I get the following exception in the second line ..
>>>>>
>>>>> org.apache.flink.api.common.functions.InvalidTypesException: The
>>>>>> return type of function 'Custom Source' could not be determined
>>>>>> automatically, due to type erasure. You can give type information hints
>>>>>> by
>>>>>> using the returns(...) method on the result of the transformation call,
>>>>>> or
>>>>>> by letting your function implement the 'ResultTypeQueryable' interface.
>>>>>
>>>>>
>>>>> Initially the returns call was not there and I was getting the same
>>>>> exception. Now after adding the returns call, nothing changes.
>>>>>
>>>>> Any help will be appreciated ..
>>>>>
>>>>> regards.
>>>>>
>>>>> --
>>>>> Debasish Ghosh
>>>>> http://manning.com/ghosh2
>>>>> http://manning.com/ghosh
>>>>>
>>>>> Twttr: @debasishg
>>>>> Blog: http://debasishg.blogspot.com
>>>>> Code: http://github.com/debasishg
>>>>>
>>>>
>>>
>>> --
>>> Debasish Ghosh
>>> http://manning.com/ghosh2
>>> http://manning.com/ghosh
>>>
>>> Twttr: @debasishg
>>> Blog: http://debasishg.blogspot.com
>>> Code: http://github.com/debasishg
>>>
>>
>
> --
> Debasish Ghosh
> http://manning.com/ghosh2
> http://manning.com/ghosh
>
> Twttr: @debasishg
> Blog: http://debasishg.blogspot.com
> Code: http://github.com/debasishg
>
--
Debasish Ghosh
http://manning.com/ghosh2
http://manning.com/ghosh
Twttr: @debasishg
Blog: http://debasishg.blogspot.com
Code: http://github.com/debasishg
