Spark转Flink SQL创建MAP报错，请求正确实现方案指导

我有一段可正常运行的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 Table API的Flink SQL，仅将IFNULL替换为COALESCE，转换后的代码如下：

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";

但运行时出现如下错误：

Caused by: org.apache.flink.client.program.ProgramInvocationException:
The main method caused an error: SQL parse failed. Non-query
expression encountered in illegal context

请问当前在Flink SQL中创建MAP的方式是否正确？请求提供解决建议。

你当前的MAP创建方式在Flink SQL中是错误的，Flink SQL的MAP构造语法和Spark SQL存在差异：

错误原因 #

Spark SQL支持MAP(key1, value1, key2, value2...)这种键值对交替传入的构造方式，但Flink SQL不支持该语法，这直接导致了SQL解析失败。

正确写法 #

根据你使用的Flink版本，有两种可行的写法：

1. 基于数组的MAP构造（兼容所有Flink版本） #

Flink SQL标准的MAP构造需要传入两个数组：一个是键数组，一个是对应的值数组，语法为MAP(ARRAY[key1, key2, ...], ARRAY[value1, value2, ...])。

修改后的代码如下：

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

2. 键值对赋值语法（Flink 1.13及以上版本支持） #

从Flink 1.13开始，支持更直观的键值对赋值语法，使用:=分隔键和值，语法为MAP(key1 := value1, key2 := value2, ...)。

修改后的代码如下：

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";

验证注意事项 #

• 确保所有键的类型一致，所有值的类型一致（Flink要求MAP的键必须是同一类型，值也必须是同一类型）。
• 如果UDF返回值类型和其他值类型不同，需要通过CAST(UDF1(param1) AS ...)做类型转换来统一。

内容的提问来源于stack exchange，提问作者stillLearning