创建 MAP 的 Flink Sql 查询

Question

我有一个采用这种格式的有效 Spark SQL 查询：

String flinkSqlQuery =
    "SELECT " +
    "  MAP( " +
    "    'mId', memberId, " +
    "    'rId', IFNULL(readId, ''), " +
    "    'VIEWED', 1, " +
    "    'UDF1', UDF1(param1) " +
    "  ) AS requestContext, " +
    "  CAST(1.0 AS FLOAT) AS FIXVAL, " +
    "  CAST(0.0 AS FLOAT) AS RESPONSE " +
    "FROM testingTable";

我正在尝试将其转换为 Flink Sql 以使用 Flink 的 Table API 运行：

String flinkSqlQuery =
    "SELECT " +
    "  MAP( " +
    "    'mId', memberId, " +
    "    'rId', COALESCE(readId, ''), " +
    "    'VIEWED', 1, " +
    "    'UDF1', UDF1(param1) " +
    "  ) AS requestContext, " +
    "  CAST(1.0 AS FLOAT) AS FIXVAL, " +
    "  CAST(0.0 AS FLOAT) AS RESPONSE " +
    "FROM testingTable";

但是我遇到了以下错误：

引起的：org.apache.flink.client.program.ProgramInitationException： main方法引发错误：SQL解析失败。非查询 在非法上下文中遇到的表达

请建议这是否是在 Flink Sql 中创建 MAP 的正确方法。

Answer 1

Flink SQL 用于创建映射（或数组）的语法涉及方括号而不是圆括号。所以应该是这样的：

    String flinkSqlQuery =
        "SELECT " +
        "  MAP[ " +
        "    'mId', memberId, " +
        "    'rId', IFNULL(readId, ''), " +
        "    'VIEWED', 1, " +
        "    'UDF1', UDF1(param1) " +
        "  ] AS requestContext, " +
        "  CAST(1.0 AS FLOAT) AS FIXVAL, " +
        "  CAST(0.0 AS FLOAT) AS RESPONSE " +
        "FROM testingTable";

请参阅文档了解更多信息。