apache-flink - 将 ROW() 用于嵌套数据结构
问题描述
我已经成功地使用来自 flink-json 工件的 JsonRowSerializationSchema 来TableSink<Row>
使用 ROW 从 SQL 创建和输出 json。它非常适合发出平面数据:
INSERT INTO outputTable
SELECT
ROW(col1, col1)
FROM inputTable
>>>> OK:
{"outCol1":"dasdasdas","outCol2":"dasdasdas"}
现在,我正在尝试一个嵌套模式,它以一种奇怪的方式分解:
INSERT INTO outputTable
SELECT
ROW('ttt', ROW('ppp'))
FROM inputTable
>>>> OK:
{"outCol1":"ttt","outCol2":{"outCol3":"ppp"}}
INSERT INTO outputTable
SELECT
ROW('ttt', ROW(col1))
FROM inputTable
>>>> OK:
{"outCol1":"ttt","outCol2":{"outCol3":"dasdasdas"}}
INSERT INTO outputTable
SELECT
ROW(col1, ROW(col1))
FROM inputTable
>>>> KO
这是一个解析问题,但我对它为什么会发生感到困惑。col1 和 'ttt' 是 String 类型的表达式,应该是可替换的;但不知何故,解析器受到以下行的干扰,正如堆栈跟踪所说:
Caused by: org.apache.calcite.sql.parser.impl.ParseException: Encountered ", ROW" at line 3, column 11.
Was expecting one of:
")" ...
"," <IDENTIFIER> ...
"," <QUOTED_IDENTIFIER> ...
"," <BACK_QUOTED_IDENTIFIER> ...
"," <BRACKET_QUOTED_IDENTIFIER> ...
"," <UNICODE_QUOTED_IDENTIFIER> ...
at org.apache.calcite.sql.parser.impl.SqlParserImpl.generateParseException(SqlParserImpl.java:23019)
at org.apache.calcite.sql.parser.impl.SqlParserImpl.jj_consume_token(SqlParserImpl.java:22836)
at org.apache.calcite.sql.parser.impl.SqlParserImpl.ParenthesizedSimpleIdentifierList(SqlParserImpl.java:4466)
at org.apache.calcite.sql.parser.impl.SqlParserImpl.Expression3(SqlParserImpl.java:3328)
at org.apache.calcite.sql.parser.impl.SqlParserImpl.Expression2b(SqlParserImpl.java:3066)
at org.apache.calcite.sql.parser.impl.SqlParserImpl.Expression2(SqlParserImpl.java:3092)
at org.apache.calcite.sql.parser.impl.SqlParserImpl.Expression(SqlParserImpl.java:3045)
at ...
我错过了一些关于语法的东西吗?解析器试图做什么?我应该以另一种方式使用 ROW() 吗?
这是一个错误吗?
解决方案
经过进一步挖掘,我得出以下结果:您只需与 ROW() 很好地交谈。
这将起作用:
INSERT INTO outputTable
SELECT ROW(col1, col2)
FROM (
SELECT
col1,
ROW(col1, col1) as col2
FROM inputTable
) tbl2
笔记:
- 嵌套:也许 SQL 只允许一个嵌套级别。但是您可以使用多个表表达式。我的看法是,在将 SQL 语义推送到执行引擎之前,Flink 目前几乎没有改变 SQL 语义。执行计划将在单个单元中创建一个融合的 ROW(col1, ROW(col1, col1)),因此影响不大。
- ROW(col1, col1):辅助表中的 ROW(col1) 将不起作用。(它将在第一个表中独立工作)。不知道为什么。但是,嘿,当我只有一个值时,我真的需要那个吗?我可以折叠那个值。如果您在输出模式中有一些余地,这将不是问题。
我在这里提交了一个 JIRA 问题:
https://issues.apache.org/jira/projects/FLINK/issues/FLINK-11399
将相应地更新此帖子
推荐阅读
- git - VSCode“请在结帐前清理您的存储库工作树”
- kubernetes - 来自 kubernetes 中的服务的偶然 Unknownhost 异常
- python - sqlalchemy关系一对一数据重复
- php - SQLSTATE[42601]:语法错误:7 内部连接错误
- matlab - matlab中“编码器”的目的是什么?
- amazon-web-services - 将一个简单的功能部署到 amazon sagemaker
- firebase - 消息:我们无法注册默认服务工作者。注册 ServiceWorker 失败
- php - Wordpress PHP 如何使用 AJAX/JQUERY 添加加载更多功能
- algorithm - 找到包含点的最小范围的最有效算法/数据结构是什么?
- python-3.x - reverse a linked list non recursive