Flink表API

Question

我正在流媒体环境中使用Flink的表API和/或Flink的SQL支持（Flink 1.3.1、Scala 2.11）。我从一个数据流【Person】开始，Person是一个case类，看起来像：

Person(name: String, age: Int, attributes: Map[String, String])

在我开始将属性带入图片之前，一切都按预期进行。

例如：

val result = html" target="_blank">streamTableEnvironment.sql(
"""
|SELECT
|name,
|attributes['foo'],
|TUMBLE_START(rowtime, INTERVAL '1' MINUTE)
|FROM myTable
|GROUP BY TUMBLE(rowtime, INTERVAL '1' MINUTE), name, attributes['foo']
|""".stripMargin)

。。。导致：

线程“main”组织中出现异常。阿帕奇。Flink。桌子api。TableException：不支持类型：组织中的任何。阿帕奇。Flink。桌子api。TableException$。在组织中应用（例外。scala:53）。阿帕奇。Flink。桌子方解石FlinkTypeFactory美元。toTypeInfo（FlinkTypeFactory.scala:341）位于org。阿帕奇。Flink。桌子计划必然的LogicalRelNode$$anonfun$12。在组织中应用（operators.scala:531）。阿帕奇。Flink。桌子计划必然的LogicalRelNode$$anonfun$12。在scala上应用（运算符。scala:530）。收集TraversableLike$$anonfun$映射$1。在scala上应用（TraversableLike.scala:234）。收集TraversableLike$$anonfun$映射$1。在scala上应用（TraversableLike.scala:234）。收集迭代器$类。scala上的foreach（迭代器。scala:893）。收集抽象迭代器。scala的foreach（迭代器。scala:1336）。收集IterableLike$类。scala的foreach（IterableLike.scala:72）。收集可抽象。scala的foreach（Iterable.scala:54）。收集TraversableLike$类。在scala处映射（TraversableLike.scala:234）。收集可抽象遍历。地图（可遍历。scala:104）位于组织。阿帕奇。Flink。桌子计划必然的LogicalRelNode。（operators.scala:530）位于组织。阿帕奇。Flink。桌子api。TableEnvironment。sql（TableEnvironment.scala:503）位于com。诺德斯特罗姆。mdt。工作$。com上的main（Job.scala:112）。诺德斯特罗姆。mdt。工作主（作业.scala）

注意：无论是否存在特定的地图键，都会发生此错误。另请注意，如果我根本没有指定地图键，我会得到一个不同的错误，这是有意义的；这种情况在这里不起作用。

这个PR似乎说有一条前进的道路：https://github.com/apache/flink/pull/3767.具体看一下测试用例，它表明类型信息可以使用DataSets。来自DataStream和的相关方法ynsterDataStream都没有提供提供类型信息的方法。

这可能吗？换句话说，流上的Flink SQL能支持映射吗？

澄清编辑...当省略map键（GROUP BY...属性而不是属性['foo']）时，我得到了下面的错误。这表明运行时确实知道这些是字符串。

此类型（接口scala.collection.immutable.Map[scala. Tuple2（_1： String，_2： String）]）不能用作键。




Answer 1

匿名用户

目前，Flink SQL只支持Java。util。地图。Scala映射被视为具有Flink数据类型的黑盒。因此，您可以转发这些黑盒并在标量函数中使用它们，但不支持使用['key']操作符进行访问。

因此，要么使用Java映射，要么自己在UDF中实现访问操作。

我为您的问题创建了一个问题：https://issues.apache.org/jira/browse/FLINK-7360