[
https://issues.apache.org/jira/browse/FLINK-10763?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16673214#comment-16673214
]
ASF GitHub Bot commented on FLINK-10763:
----------------------------------------
manbuyun opened a new pull request #7007: [FLINK-10763][streaming] Cannot union
streams of different types in the union method of DataStream class
URL: https://github.com/apache/flink/pull/7007
JIRA Issue: https://issues.apache.org/jira/browse/FLINK-10763
When stream is a Scala case class, the TypeInformation will fall back to
GenericType in the process function which result in bad performance when union
another DataStream
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> Cannot union streams of different types in the union method of DataStream
> class
> -------------------------------------------------------------------------------
>
> Key: FLINK-10763
> URL: https://issues.apache.org/jira/browse/FLINK-10763
> Project: Flink
> Issue Type: Bug
> Components: DataStream API
> Affects Versions: 1.3.3, 1.4.2, 1.5.5, 1.6.2
> Reporter: wangjinhai
> Priority: Major
> Labels: pull-request-available
> Fix For: 1.5.6, 1.6.3, 1.7.0
>
>
> When stream is a Scala case class, the TypeInformation will fall back to
> GenericType in the process function which result in bad performance when
> union another DataStream.
> In the union method of DataStream, the type is first checked for equality.
> Here is an example:
> {code:java}
> object Test {
> def main(args: Array[String]): Unit = {
> val env = StreamExecutionEnvironment.getExecutionEnvironment
> val orderA: DataStream[Order] = env.fromCollection(Seq(
> Order(1L, "beer", 3),
> Order(1L, "diaper", 4),
> Order(3L, "rubber", 2)))
> val orderB: DataStream[Order] = env.fromCollection(Seq(
> new Order(2L, "pen", 3),
> new Order(2L, "rubber", 3),
> new Order(4L, "beer", 1)))
> val orderC: DataStream[Order] = orderA.keyBy(_.user)
> .intervalJoin(orderB.keyBy(_.user))
> .between(Time.seconds(0), Time.seconds(0))
> .process(new ProcessJoinFunction[Order, Order, Order] {
> override def processElement(left: Order, right: Order, ctx:
> ProcessJoinFunction[Order, Order, Order]#Context, out: Collector[Order]):
> Unit = {
> out.collect(left)
> }})
> println("C: " + orderC.dataType.toString)
> println("B: " + orderB.dataType.toString)
> orderC.union(orderB).print()
> env.execute()
> }
> case class Order(user: Long, product: String, amount: Int)
> }{code}
> Here is the Exception:
> {code:java}
> Exception in thread "main" java.lang.IllegalArgumentException: Cannot union
> streams of different types:
> GenericType<com.manbuyun.awesome.flink.Test.Order> and
> com.manbuyun.awesome.flink.Test$Order(user: Long, product: String, amount:
> Integer)
> at
> org.apache.flink.streaming.api.datastream.DataStream.union(DataStream.java:219)
> at
> org.apache.flink.streaming.api.scala.DataStream.union(DataStream.scala:357)
> at com.manbuyun.awesome.flink.Test$.main(Test.scala:38)
> at com.manbuyun.awesome.flink.Test.main(Test.scala){code}
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
