You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by YI <uu...@protonmail.com> on 2020/06/18 02:32:16 UTC
Convert flink table with field of type RAW to datastream
Hi all,
I am using flink to process external data. The source format is json, and the underlying data types are defined in a external library.
I generated table schema with `TableSchema.fromTypeInfo` and `TypeInformation.of[_]`. From what I read, this method is deprecated.
But I didn't find any alternatives. Manually tweaking table schema is not viable as there are simply too many types.
One of the field in the source type is `java.util.Date`. I tried to convert the obtained table to a datastream with Table.toAppendStream.
When I ran `tEnv.from("rawEvent").select('_isComplete).toAppendStream[(Boolean)].print()`, the following exception occurred.
Exception in thread "main" org.apache.flink.table.api.TableException: Type is not supported: Date
at org.apache.flink.table.calcite.FlinkTypeFactory$.org$apache$flink$table$calcite$FlinkTypeFactory$$typeInfoToSqlTypeName(FlinkTypeFactory.scala:350)
at org.apache.flink.table.calcite.FlinkTypeFactory.createTypeFromTypeInfo(FlinkTypeFactory.scala:63)
at org.apache.flink.table.calcite.FlinkTypeFactory.$anonfun$buildLogicalRowType$1(FlinkTypeFactory.scala:201)
at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
at org.apache.flink.table.calcite.FlinkTypeFactory.buildLogicalRowType(FlinkTypeFactory.scala:198)
at org.apache.flink.table.plan.schema.TableSourceTable.getRowType(TableSourceTable.scala:96)
at org.apache.calcite.prepare.CalciteCatalogReader.getTable(CalciteCatalogReader.java:131)
at org.apache.calcite.prepare.CalciteCatalogReader.getTableForMember(CalciteCatalogReader.java:228)
at org.apache.calcite.prepare.CalciteCatalogReader.getTableForMember(CalciteCatalogReader.java:84)
at org.apache.calcite.tools.RelBuilder.scan(RelBuilder.java:1068)
at org.apache.calcite.tools.RelBuilder.scan(RelBuilder.java:1094)
at org.apache.flink.table.plan.QueryOperationConverter$SingleRelVisitor.visit(QueryOperationConverter.java:268)
at org.apache.flink.table.plan.QueryOperationConverter$SingleRelVisitor.visit(QueryOperationConverter.java:134)
at org.apache.flink.table.operations.CatalogQueryOperation.accept(CatalogQueryOperation.java:69)
at org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:131)
at org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:111)
at org.apache.flink.table.operations.utils.QueryOperationDefaultVisitor.visit(QueryOperationDefaultVisitor.java:91)
at org.apache.flink.table.operations.CatalogQueryOperation.accept(CatalogQueryOperation.java:69)
at org.apache.flink.table.plan.QueryOperationConverter.lambda$defaultMethod$0(QueryOperationConverter.java:130)
at java.util.Collections$SingletonList.forEach(Collections.java:4824)
at org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:130)
at org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:111)
at org.apache.flink.table.operations.utils.QueryOperationDefaultVisitor.visit(QueryOperationDefaultVisitor.java:46)
at org.apache.flink.table.operations.ProjectQueryOperation.accept(ProjectQueryOperation.java:75)
at org.apache.flink.table.calcite.FlinkRelBuilder.tableOperation(FlinkRelBuilder.scala:106)
at org.apache.flink.table.planner.StreamPlanner.translateToType(StreamPlanner.scala:390)
at org.apache.flink.table.planner.StreamPlanner.translate(StreamPlanner.scala:185)
at org.apache.flink.table.planner.StreamPlanner.$anonfun$translate$1(StreamPlanner.scala:117)
at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:273)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
at scala.collection.TraversableLike.map(TraversableLike.scala:273)
at scala.collection.TraversableLike.map$(TraversableLike.scala:266)
at scala.collection.AbstractTraversable.map(Traversable.scala:108)
at org.apache.flink.table.planner.StreamPlanner.translate(StreamPlanner.scala:117)
at org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.toDataStream(StreamTableEnvironmentImpl.scala:210)
at org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.toAppendStream(StreamTableEnvironmentImpl.scala:107)
at org.apache.flink.table.api.scala.TableConversions.toAppendStream(TableConversions.scala:101)
at io.redacted.test.package$.testJoin(package.scala:31)
at io.redacted.test.package$.process(package.scala:26)
at io.redacted.DataAggregator$.main(DataAggregator.scala:15)
at io.redacted.DataAggregator.main(DataAggregator.scala)
This exception is thrown even though I didn't select RAW data field `_startTime` which is of type `java.util.Date`. I believe this exception is undesirable.
Is there any way to obtain a RAW data from flink tables? If there isn't any, how do I circumvent my current issue? Do I need to manually update all table schema?
There is a relevant issue in http://mail-archives.apache.org/mod_mbox/flink-user/201907.mbox/%3CCA+3UsY2-L1OKTjNBwX2ajG3o6v5M6QS=jBwyYbeMZLVDM5xgXg@mail.gmail.com%3E,
Unfortunately, I didn't find a satisfatory solutions.
Cheers,
Yi
Re: Convert flink table with field of type RAW to datastream
Posted by Jark Wu <im...@gmail.com>.
Flink SQL/Table requires to know the field data types explicitly. Maybe you
can apply a MapFunction before `toTable` to convert/normalize the data and
type.
Best,
Jark
On Thu, 18 Jun 2020 at 14:12, YI <uu...@protonmail.com> wrote:
> Hi Jark,
>
> Thank you for your suggestion. My current problem is that there are quite
> a few data types. All these data types are defined upstream which I have no
> control.
> I don't think I can easily change the type information of a specific
> field. Can I? Things become nasty when there are so many `java.util.Date` I
> need to change.
>
> The reason I want to use flink table is that it allows me to easily join
> several tables. As an alternative, I think I can use stream join operator.
> My only complaint is that it become tedious when I want to join more than
> once. I think I need to define all the intermediate data types.
>
> Best,
> Yi
>
>
>
> ‐‐‐‐‐‐‐ Original Message ‐‐‐‐‐‐‐
> On Thursday, June 18, 2020 12:11 PM, Jark Wu <im...@gmail.com> wrote:
>
> Hi YI,
>
> Flink doesn't have a TypeInformation for `java.util.Date`, but
> only SqlTimeTypeInfo.DATE for `java.sql.Date`.
> That's why the TypeInformation.of(java.util.Date) is being recognized as a
> RAW type.
>
> To resolve your problem, I think in `TypeInformation.of(..)` you should
> use a concrete type for `java.util.Date`, e.g. `java.sql.Timestamp`,
> `java.sql.Date`, `java.sql.Time`.
>
> Best,
> Jark
>
> On Thu, 18 Jun 2020 at 10:32, YI <uu...@protonmail.com> wrote:
>
>> Hi all,
>>
>> I am using flink to process external data. The source format is json, and
>> the underlying data types are defined in a external library.
>> I generated table schema with `TableSchema.fromTypeInfo` and
>> `TypeInformation.of[_]`. From what I read, this method is deprecated.
>> But I didn't find any alternatives. Manually tweaking table schema is not
>> viable as there are simply too many types.
>>
>> One of the field in the source type is `java.util.Date`. I tried to
>> convert the obtained table to a datastream with Table.toAppendStream.
>> When I ran
>> `tEnv.from("rawEvent").select('_isComplete).toAppendStream[(Boolean)].print()`,
>> the following exception occurred.
>>
>> Exception in thread "main" org.apache.flink.table.api.TableException:
>> Type is not supported: Date
>> at
>> org.apache.flink.table.calcite.FlinkTypeFactory$.org$apache$flink$table$calcite$FlinkTypeFactory$$typeInfoToSqlTypeName(FlinkTypeFactory.scala:350)
>> at
>> org.apache.flink.table.calcite.FlinkTypeFactory.createTypeFromTypeInfo(FlinkTypeFactory.scala:63)
>> at
>> org.apache.flink.table.calcite.FlinkTypeFactory.$anonfun$buildLogicalRowType$1(FlinkTypeFactory.scala:201)
>> at
>> scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
>> at
>> scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
>> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
>> at
>> org.apache.flink.table.calcite.FlinkTypeFactory.buildLogicalRowType(FlinkTypeFactory.scala:198)
>> at
>> org.apache.flink.table.plan.schema.TableSourceTable.getRowType(TableSourceTable.scala:96)
>> at
>> org.apache.calcite.prepare.CalciteCatalogReader.getTable(CalciteCatalogReader.java:131)
>> at
>> org.apache.calcite.prepare.CalciteCatalogReader.getTableForMember(CalciteCatalogReader.java:228)
>> at
>> org.apache.calcite.prepare.CalciteCatalogReader.getTableForMember(CalciteCatalogReader.java:84)
>> at org.apache.calcite.tools.RelBuilder.scan(RelBuilder.java:1068)
>> at org.apache.calcite.tools.RelBuilder.scan(RelBuilder.java:1094)
>> at
>> org.apache.flink.table.plan.QueryOperationConverter$SingleRelVisitor.visit(QueryOperationConverter.java:268)
>> at
>> org.apache.flink.table.plan.QueryOperationConverter$SingleRelVisitor.visit(QueryOperationConverter.java:134)
>> at
>> org.apache.flink.table.operations.CatalogQueryOperation.accept(CatalogQueryOperation.java:69)
>> at
>> org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:131)
>> at
>> org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:111)
>> at
>> org.apache.flink.table.operations.utils.QueryOperationDefaultVisitor.visit(QueryOperationDefaultVisitor.java:91)
>> at
>> org.apache.flink.table.operations.CatalogQueryOperation.accept(CatalogQueryOperation.java:69)
>> at
>> org.apache.flink.table.plan.QueryOperationConverter.lambda$defaultMethod$0(QueryOperationConverter.java:130)
>> at java.util.Collections$SingletonList.forEach(Collections.java:4824)
>> at
>> org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:130)
>> at
>> org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:111)
>> at
>> org.apache.flink.table.operations.utils.QueryOperationDefaultVisitor.visit(QueryOperationDefaultVisitor.java:46)
>> at
>> org.apache.flink.table.operations.ProjectQueryOperation.accept(ProjectQueryOperation.java:75)
>> at
>> org.apache.flink.table.calcite.FlinkRelBuilder.tableOperation(FlinkRelBuilder.scala:106)
>> at
>> org.apache.flink.table.planner.StreamPlanner.translateToType(StreamPlanner.scala:390)
>> at
>> org.apache.flink.table.planner.StreamPlanner.translate(StreamPlanner.scala:185)
>> at
>> org.apache.flink.table.planner.StreamPlanner.$anonfun$translate$1(StreamPlanner.scala:117)
>> at
>> scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:273)
>> at scala.collection.Iterator.foreach(Iterator.scala:943)
>> at scala.collection.Iterator.foreach$(Iterator.scala:943)
>> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
>> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
>> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
>> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
>> at scala.collection.TraversableLike.map(TraversableLike.scala:273)
>> at scala.collection.TraversableLike.map$(TraversableLike.scala:266)
>> at scala.collection.AbstractTraversable.map(Traversable.scala:108)
>> at
>> org.apache.flink.table.planner.StreamPlanner.translate(StreamPlanner.scala:117)
>> at
>> org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.toDataStream(StreamTableEnvironmentImpl.scala:210)
>> at
>> org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.toAppendStream(StreamTableEnvironmentImpl.scala:107)
>> at
>> org.apache.flink.table.api.scala.TableConversions.toAppendStream(TableConversions.scala:101)
>> at io.redacted.test.package$.testJoin(package.scala:31)
>> at io.redacted.test.package$.process(package.scala:26)
>> at io.redacted.DataAggregator$.main(DataAggregator.scala:15)
>> at io.redacted.DataAggregator.main(DataAggregator.scala)
>>
>>
>> This exception is thrown even though I didn't select RAW data field
>> `_startTime` which is of type `java.util.Date`. I believe this exception is
>> undesirable.
>> Is there any way to obtain a RAW data from flink tables? If there isn't
>> any, how do I circumvent my current issue? Do I need to manually update all
>> table schema?
>>
>> There is a relevant issue in
>> http://mail-archives.apache.org/mod_mbox/flink-user/201907.mbox/%3CCA+3UsY2-L1OKTjNBwX2ajG3o6v5M6QS=jBwyYbeMZLVDM5xgXg@mail.gmail.com%3E
>> ,
>> Unfortunately, I didn't find a satisfatory solutions.
>>
>> Cheers,
>> Yi
>>
>>
>
Re: Convert flink table with field of type RAW to datastream
Posted by YI <uu...@protonmail.com>.
Hi Jark,
Thank you for your suggestion. My current problem is that there are quite a few data types. All these data types are defined upstream which I have no control.
I don't think I can easily change the type information of a specific field. Can I? Things become nasty when there are so many `java.util.Date` I need to change.
The reason I want to use flink table is that it allows me to easily join several tables. As an alternative, I think I can use stream join operator.
My only complaint is that it become tedious when I want to join more than once. I think I need to define all the intermediate data types.
Best,
Yi
‐‐‐‐‐‐‐ Original Message ‐‐‐‐‐‐‐
On Thursday, June 18, 2020 12:11 PM, Jark Wu <im...@gmail.com> wrote:
> Hi YI,
>
> Flink doesn't have a TypeInformation for `java.util.Date`, but only SqlTimeTypeInfo.DATE for `java.sql.Date`.
> That's why the TypeInformation.of(java.util.Date) is being recognized as a RAW type.
>
> To resolve your problem, I think in `TypeInformation.of(..)` you should use a concrete type for `java.util.Date`, e.g. `java.sql.Timestamp`, `java.sql.Date`, `java.sql.Time`.
>
> Best,
> Jark
>
> On Thu, 18 Jun 2020 at 10:32, YI <uu...@protonmail.com> wrote:
>
>> Hi all,
>>
>> I am using flink to process external data. The source format is json, and the underlying data types are defined in a external library.
>> I generated table schema with `TableSchema.fromTypeInfo` and `TypeInformation.of[_]`. From what I read, this method is deprecated.
>> But I didn't find any alternatives. Manually tweaking table schema is not viable as there are simply too many types.
>>
>> One of the field in the source type is `java.util.Date`. I tried to convert the obtained table to a datastream with Table.toAppendStream.
>> When I ran `tEnv.from("rawEvent").select('_isComplete).toAppendStream[(Boolean)].print()`, the following exception occurred.
>>
>> Exception in thread "main" org.apache.flink.table.api.TableException: Type is not supported: Date
>> at org.apache.flink.table.calcite.FlinkTypeFactory$.org$apache$flink$table$calcite$FlinkTypeFactory$$typeInfoToSqlTypeName(FlinkTypeFactory.scala:350)
>> at org.apache.flink.table.calcite.FlinkTypeFactory.createTypeFromTypeInfo(FlinkTypeFactory.scala:63)
>> at org.apache.flink.table.calcite.FlinkTypeFactory.$anonfun$buildLogicalRowType$1(FlinkTypeFactory.scala:201)
>> at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
>> at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
>> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
>> at org.apache.flink.table.calcite.FlinkTypeFactory.buildLogicalRowType(FlinkTypeFactory.scala:198)
>> at org.apache.flink.table.plan.schema.TableSourceTable.getRowType(TableSourceTable.scala:96)
>> at org.apache.calcite.prepare.CalciteCatalogReader.getTable(CalciteCatalogReader.java:131)
>> at org.apache.calcite.prepare.CalciteCatalogReader.getTableForMember(CalciteCatalogReader.java:228)
>> at org.apache.calcite.prepare.CalciteCatalogReader.getTableForMember(CalciteCatalogReader.java:84)
>> at org.apache.calcite.tools.RelBuilder.scan(RelBuilder.java:1068)
>> at org.apache.calcite.tools.RelBuilder.scan(RelBuilder.java:1094)
>> at org.apache.flink.table.plan.QueryOperationConverter$SingleRelVisitor.visit(QueryOperationConverter.java:268)
>> at org.apache.flink.table.plan.QueryOperationConverter$SingleRelVisitor.visit(QueryOperationConverter.java:134)
>> at org.apache.flink.table.operations.CatalogQueryOperation.accept(CatalogQueryOperation.java:69)
>> at org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:131)
>> at org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:111)
>> at org.apache.flink.table.operations.utils.QueryOperationDefaultVisitor.visit(QueryOperationDefaultVisitor.java:91)
>> at org.apache.flink.table.operations.CatalogQueryOperation.accept(CatalogQueryOperation.java:69)
>> at org.apache.flink.table.plan.QueryOperationConverter.lambda$defaultMethod$0(QueryOperationConverter.java:130)
>> at java.util.Collections$SingletonList.forEach(Collections.java:4824)
>> at org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:130)
>> at org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:111)
>> at org.apache.flink.table.operations.utils.QueryOperationDefaultVisitor.visit(QueryOperationDefaultVisitor.java:46)
>> at org.apache.flink.table.operations.ProjectQueryOperation.accept(ProjectQueryOperation.java:75)
>> at org.apache.flink.table.calcite.FlinkRelBuilder.tableOperation(FlinkRelBuilder.scala:106)
>> at org.apache.flink.table.planner.StreamPlanner.translateToType(StreamPlanner.scala:390)
>> at org.apache.flink.table.planner.StreamPlanner.translate(StreamPlanner.scala:185)
>> at org.apache.flink.table.planner.StreamPlanner.$anonfun$translate$1(StreamPlanner.scala:117)
>> at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:273)
>> at scala.collection.Iterator.foreach(Iterator.scala:943)
>> at scala.collection.Iterator.foreach$(Iterator.scala:943)
>> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
>> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
>> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
>> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
>> at scala.collection.TraversableLike.map(TraversableLike.scala:273)
>> at scala.collection.TraversableLike.map$(TraversableLike.scala:266)
>> at scala.collection.AbstractTraversable.map(Traversable.scala:108)
>> at org.apache.flink.table.planner.StreamPlanner.translate(StreamPlanner.scala:117)
>> at org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.toDataStream(StreamTableEnvironmentImpl.scala:210)
>> at org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.toAppendStream(StreamTableEnvironmentImpl.scala:107)
>> at org.apache.flink.table.api.scala.TableConversions.toAppendStream(TableConversions.scala:101)
>> at io.redacted.test.package$.testJoin(package.scala:31)
>> at io.redacted.test.package$.process(package.scala:26)
>> at io.redacted.DataAggregator$.main(DataAggregator.scala:15)
>> at io.redacted.DataAggregator.main(DataAggregator.scala)
>>
>> This exception is thrown even though I didn't select RAW data field `_startTime` which is of type `java.util.Date`. I believe this exception is undesirable.
>> Is there any way to obtain a RAW data from flink tables? If there isn't any, how do I circumvent my current issue? Do I need to manually update all table schema?
>>
>> There is a relevant issue in http://mail-archives.apache.org/mod_mbox/flink-user/201907.mbox/%3CCA+3UsY2-L1OKTjNBwX2ajG3o6v5M6QS=jBwyYbeMZLVDM5xgXg@mail.gmail.com%3E,
>> Unfortunately, I didn't find a satisfatory solutions.
>>
>> Cheers,
>> Yi
Re: Convert flink table with field of type RAW to datastream
Posted by Jark Wu <im...@gmail.com>.
Hi YI,
Flink doesn't have a TypeInformation for `java.util.Date`, but
only SqlTimeTypeInfo.DATE for `java.sql.Date`.
That's why the TypeInformation.of(java.util.Date) is being recognized as a
RAW type.
To resolve your problem, I think in `TypeInformation.of(..)` you should use
a concrete type for `java.util.Date`, e.g. `java.sql.Timestamp`,
`java.sql.Date`, `java.sql.Time`.
Best,
Jark
On Thu, 18 Jun 2020 at 10:32, YI <uu...@protonmail.com> wrote:
> Hi all,
>
> I am using flink to process external data. The source format is json, and
> the underlying data types are defined in a external library.
> I generated table schema with `TableSchema.fromTypeInfo` and
> `TypeInformation.of[_]`. From what I read, this method is deprecated.
> But I didn't find any alternatives. Manually tweaking table schema is not
> viable as there are simply too many types.
>
> One of the field in the source type is `java.util.Date`. I tried to
> convert the obtained table to a datastream with Table.toAppendStream.
> When I ran
> `tEnv.from("rawEvent").select('_isComplete).toAppendStream[(Boolean)].print()`,
> the following exception occurred.
>
> Exception in thread "main" org.apache.flink.table.api.TableException: Type
> is not supported: Date
> at
> org.apache.flink.table.calcite.FlinkTypeFactory$.org$apache$flink$table$calcite$FlinkTypeFactory$$typeInfoToSqlTypeName(FlinkTypeFactory.scala:350)
> at
> org.apache.flink.table.calcite.FlinkTypeFactory.createTypeFromTypeInfo(FlinkTypeFactory.scala:63)
> at
> org.apache.flink.table.calcite.FlinkTypeFactory.$anonfun$buildLogicalRowType$1(FlinkTypeFactory.scala:201)
> at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
> at
> scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
> at
> org.apache.flink.table.calcite.FlinkTypeFactory.buildLogicalRowType(FlinkTypeFactory.scala:198)
> at
> org.apache.flink.table.plan.schema.TableSourceTable.getRowType(TableSourceTable.scala:96)
> at
> org.apache.calcite.prepare.CalciteCatalogReader.getTable(CalciteCatalogReader.java:131)
> at
> org.apache.calcite.prepare.CalciteCatalogReader.getTableForMember(CalciteCatalogReader.java:228)
> at
> org.apache.calcite.prepare.CalciteCatalogReader.getTableForMember(CalciteCatalogReader.java:84)
> at org.apache.calcite.tools.RelBuilder.scan(RelBuilder.java:1068)
> at org.apache.calcite.tools.RelBuilder.scan(RelBuilder.java:1094)
> at
> org.apache.flink.table.plan.QueryOperationConverter$SingleRelVisitor.visit(QueryOperationConverter.java:268)
> at
> org.apache.flink.table.plan.QueryOperationConverter$SingleRelVisitor.visit(QueryOperationConverter.java:134)
> at
> org.apache.flink.table.operations.CatalogQueryOperation.accept(CatalogQueryOperation.java:69)
> at
> org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:131)
> at
> org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:111)
> at
> org.apache.flink.table.operations.utils.QueryOperationDefaultVisitor.visit(QueryOperationDefaultVisitor.java:91)
> at
> org.apache.flink.table.operations.CatalogQueryOperation.accept(CatalogQueryOperation.java:69)
> at
> org.apache.flink.table.plan.QueryOperationConverter.lambda$defaultMethod$0(QueryOperationConverter.java:130)
> at java.util.Collections$SingletonList.forEach(Collections.java:4824)
> at
> org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:130)
> at
> org.apache.flink.table.plan.QueryOperationConverter.defaultMethod(QueryOperationConverter.java:111)
> at
> org.apache.flink.table.operations.utils.QueryOperationDefaultVisitor.visit(QueryOperationDefaultVisitor.java:46)
> at
> org.apache.flink.table.operations.ProjectQueryOperation.accept(ProjectQueryOperation.java:75)
> at
> org.apache.flink.table.calcite.FlinkRelBuilder.tableOperation(FlinkRelBuilder.scala:106)
> at
> org.apache.flink.table.planner.StreamPlanner.translateToType(StreamPlanner.scala:390)
> at
> org.apache.flink.table.planner.StreamPlanner.translate(StreamPlanner.scala:185)
> at
> org.apache.flink.table.planner.StreamPlanner.$anonfun$translate$1(StreamPlanner.scala:117)
> at
> scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:273)
> at scala.collection.Iterator.foreach(Iterator.scala:943)
> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> at scala.collection.TraversableLike.map(TraversableLike.scala:273)
> at scala.collection.TraversableLike.map$(TraversableLike.scala:266)
> at scala.collection.AbstractTraversable.map(Traversable.scala:108)
> at
> org.apache.flink.table.planner.StreamPlanner.translate(StreamPlanner.scala:117)
> at
> org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.toDataStream(StreamTableEnvironmentImpl.scala:210)
> at
> org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.toAppendStream(StreamTableEnvironmentImpl.scala:107)
> at
> org.apache.flink.table.api.scala.TableConversions.toAppendStream(TableConversions.scala:101)
> at io.redacted.test.package$.testJoin(package.scala:31)
> at io.redacted.test.package$.process(package.scala:26)
> at io.redacted.DataAggregator$.main(DataAggregator.scala:15)
> at io.redacted.DataAggregator.main(DataAggregator.scala)
>
>
> This exception is thrown even though I didn't select RAW data field
> `_startTime` which is of type `java.util.Date`. I believe this exception is
> undesirable.
> Is there any way to obtain a RAW data from flink tables? If there isn't
> any, how do I circumvent my current issue? Do I need to manually update all
> table schema?
>
> There is a relevant issue in
> http://mail-archives.apache.org/mod_mbox/flink-user/201907.mbox/%3CCA+3UsY2-L1OKTjNBwX2ajG3o6v5M6QS=jBwyYbeMZLVDM5xgXg@mail.gmail.com%3E
> ,
> Unfortunately, I didn't find a satisfatory solutions.
>
> Cheers,
> Yi
>
>