对嵌套数据结构使用row()

2ledvvac 于 2021-06-21 发布在 Flink

关注(0)|答案(1)|浏览(658)

我已经成功地使用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是字符串类型的表达式，应该是可替换的；但不知何故，解析器会被下面的行所干扰，正如stacktrace所说：

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（）吗？
这是虫子吗？

apache-flink flink-sql flink-streaming apache-calcite

来源：https://stackoverflow.com/questions/54224170/using-row-for-nested-data-structure

1条答案

按热度按时间

hm2xizp91#

经过进一步的挖掘，我得到了以下结果：你只需要和row（）好好谈谈。
这将起作用：

INSERT INTO outputTable
SELECT ROW(col1, col2) 
FROM (
  SELECT 
    col1, 
    ROW(col1, col1) as col2 
  FROM inputTable
) tbl2

注：
嵌套：也许sql只允许一个嵌套级别。但是允许您使用多个表表达式。我的看法是，在将sql语义推送到执行引擎之前，flink目前很少转换sql语义。执行计划将在单个单元中创建一个融合的行（col1，row（col1，col1）），因此这没有那么大的影响。
行（col1，col1）：辅助表中的行（col1）将不起作用(它将在第一个表中独立工作）。不知道为什么。但是，当我只有一个值的时候，我真的需要它吗？我可以折叠这个值。如果您在输出模式中有一些回旋余地，这不会是一个问题。
我在这里提交了一个jira问题：
https://issues.apache.org/jira/projects/flink/issues/flink-11399
将相应地更新此帖子

赞(0）回复(0）举报 2021-06-21

我来回答

对嵌套数据结构使用row()

1条答案

相关问题

热门标签

最新问答