Convert flink table with field of type RAW to datastream
Locked
 
|
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@...%3E, Unfortunately, I didn't find a satisfatory solutions. Cheers, Yi |
|
|
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 <[hidden email]> 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 <[hidden email]> wrote:
|
|
|
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 <[hidden email]> wrote:
|
«
Return to (DEPRECATED) Apache Flink User Mailing List archive.
|
1 view|%1 views
| Free forum by Nabble | Edit this page |