hequn8128 commented on a change in pull request #6787: [FLINK-8577][table]
Implement proctime DataStream to Table upsert conversion
URL: https://github.com/apache/flink/pull/6787#discussion_r236266862
##########
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/DataStreamConversions.scala
##########
@@ -59,5 +59,38 @@ class DataStreamConversions[T](dataStream: DataStream[T],
inputType: TypeInforma
}
}
+ /**
+ * Converts the [[DataStream]] with upsert messages into a [[Table]] with
keys.
+ *
+ * The message will be encoded as [[Tuple2]]. The first field is a
[[Boolean]] flag, the second
+ * field holds the record. A true [[Boolean]] flag indicates an update
message, a false flag
+ * indicates a delete message.
+ *
+ * The field name and key of the new [[Table]] can be specified like this:
+ *
+ * {{{
+ * val env = StreamExecutionEnvironment.getExecutionEnvironment
+ * val tEnv = TableEnvironment.getTableEnvironment(env)
+ *
+ * val stream: DataStream[(Boolean, (String, Int))] = ...
+ * val table = stream.toKeyedTable(tEnv, 'name.key, 'amount)
+ * }}}
+ *
+ * If field names are not explicitly specified, names are automatically
extracted from the type
+ * of the [[DataStream]].
+ * If keys are not explicitly specified, an empty key will be used and the
table will be a
+ * single row table.
+ *
+ * @param tableEnv The [[StreamTableEnvironment]] in which the new
[[Table]] is created.
+ * @param fields The field names of the new [[Table]] (optional).
+ * @return The resulting [[Table]].
+ */
+ def toKeyedTable(tableEnv: StreamTableEnvironment, fields: Expression*):
Table = {
Review comment:
I think we can't use upsert to describe a Table. How about rename it to
`toTableFromUpsert`?
----------------------------------------------------------------
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:
us...@infra.apache.org
With regards,
Apache Git Services
