Flink 的嵌套输出答案

【问题标题】：Nested output of FlinkFlink 的嵌套输出
【发布时间】：2019-08-19 13:36:44
【问题描述】：

我正在使用 Flink SQL 处理 Kafka 流，其中每条消息都从 Kafka 中提取，使用 flink sql 处理并推回 kafka。我想要一个嵌套输出，其中输入是平坦的，输出是嵌套的。比如说我的输入是

{'StudentName':'ABC','StudentAge':33}

并希望输出为

{'Student':{'Name':'ABC','Age':33}}

我尝试搜索here 和一些类似的链接，但找不到。是否可以使用 Apache Flink SQL API 来做到这一点？如有必要，可以使用用户定义的函数，但希望避免这样做。

【问题讨论】：

标签： apache-kafka apache-flink flink-streaming flink-sql

【解决方案1】：

你可以试试这样的：

SELECT 
  MAP ['Student', MAP ['Name', StudentName, 'Age', StudentAge]] 
FROM 
  students;

我找到了 MAP 函数 here，但我不得不在 SQL Client 中进行实验以找出语法。

【讨论】：

【解决方案2】：

我可以通过从 Flink UDF 返回一个地图来实现同样的效果。 UDF 中的 eval() 函数将返回一个 Map，而 FlinkSQL 查询将以学生为别名调用 UDF：

UDF 应如下所示

    public class getStudent extends ScalarFunction {
        public Map<String, String> eval(String name, Integer age) {
            Map<String, String> student = new HashMap<>();
            student.put("Name", name);
            student.put("Age", age.toString());
            return student;
        }
    }

FlinkSQL 查询保持如下：

    Select getStudent(StudentName, StudentAge) as `Student` from MyKafkaTopic

在尝试从 FlinkSQL 中获取 List 时，也可以对 List 进行同样的操作

【讨论】：