Flink：从行字段中提取数组

Question

我正在使用 Flink，并且我正在使用地图中的自定义函数。这个自定义函数使用 Flink Row 类型作为输入，并输出一个 (String, Object) 的 Map，其中包含我行的每个字段和值。

在基本情况下，此函数运行良好，但现在我需要对特定字段进行一些处理，这是一个整数数组。在这种情况下，我在将 Row 中的数据提取到 Java 集合对象（列表或数组等）时遇到了一些麻烦。

这是我CustomMap的代码：

public class CustomMap implements MapFunction<Row, Map<String, Object>> {

    private final String arrayField = "ArrayField";
    private final String[] fields = {"genTimestampMs", "some_field", "timestampMs", "some_field_2", "ArrayField"};

    public CustomMap(){}

    @Override
    public Map<String, Object> map(Row myRow) throws Exception {
        LOGGER.debug("Mapping the row "+myRow.toString());

        final Map<String, Object> m = new HashMap<>();

        for (int i = 0; i < myRow.getArity(); i++) {
            LOGGER.debug("  Field "+i);
            if (arrayField.equals(fields[i])) {
                LOGGER.debug("Is the field  "+arrayField);
                Integer wCount = 0;

                LOGGER.debug("  row0 : "+myRow.getField(i));
                Row test = Row.of(myRow.getField(i));
                LOGGER.debug("  row : "+test);
                LOGGER.debug("  getArity: "+test.getArity());

                List<Integer> myList = (List<Integer>)myRow.getField(i); // <--- Error here

                String value = // Do something with my list

                m.put(fields[i], value);

            } else {
                LOGGER.debug("  Put field in map : ("+fields[i]+" -> "+myRow.getField(i)+")");
                m.put(fields[i], myRow.getField(i));
            }
        }
        return m;
    }
}

这是我用作输入数据的 json 示例（使用 Flink JsonRowDeserializationSchema 解析）：

{"genTimestampMs": 1561130625000, "some_field": "some_value", "timestampMs": 1561130625000, "some_field_2":"some_value", "ArrayField": [1,2,3,4,5]}

这里是我对这些数据执行代码的日志：

2019-06-27 13:40:02.854 [Source: Custom Source -> Map -> Map -> Sink: Unnamed (5/12)] DEBUG CustomMap  - Mapping the row 1561130625000,some_value,1561130625000,some_value,[1, 2, 3, 4, 5]
2019-06-27 13:40:02.854 [Source: Custom Source -> Map -> Map -> Sink: Unnamed (5/12)] DEBUG CustomMap  -   Field 0
2019-06-27 13:40:02.858 [Source: Custom Source -> Map -> Map -> Sink: Unnamed (5/12)] DEBUG CustomMap  -   Put field in map : (genTimestampMs -> 1561130625000)
2019-06-27 13:40:02.858 [Source: Custom Source -> Map -> Map -> Sink: Unnamed (5/12)] DEBUG CustomMap  -   Field 1
2019-06-27 13:40:02.858 [Source: Custom Source -> Map -> Map -> Sink: Unnamed (5/12)] DEBUG CustomMap  -   Put field in map : (some_field -> some_value)
2019-06-27 13:40:02.858 [Source: Custom Source -> Map -> Map -> Sink: Unnamed (5/12)] DEBUG CustomMap  -   Field 2
2019-06-27 13:40:02.858 [Source: Custom Source -> Map -> Map -> Sink: Unnamed (5/12)] DEBUG CustomMap  -   Put field in map : (timestampMs -> 1561130625000)
2019-06-27 13:40:02.858 [Source: Custom Source -> Map -> Map -> Sink: Unnamed (5/12)] DEBUG CustomMap  -   Field 3
2019-06-27 13:40:02.858 [Source: Custom Source -> Map -> Map -> Sink: Unnamed (5/12)] DEBUG CustomMap  -   Put field in map : (some_field_2 -> some_value)
2019-06-27 13:40:02.858 [Source: Custom Source -> Map -> Map -> Sink: Unnamed (5/12)] DEBUG CustomMap  -   Field 4
2019-06-27 13:40:02.858 [Source: Custom Source -> Map -> Map -> Sink: Unnamed (5/12)] DEBUG CustomMap  -   Is the field  ArrayField
2019-06-27 13:40:02.858 [Source: Custom Source -> Map -> Map -> Sink: Unnamed (5/12)] DEBUG CustomMap  -   row0 : [Ljava.lang.Integer;@68374747
2019-06-27 13:40:02.859 [Source: Custom Source -> Map -> Map -> Sink: Unnamed (5/12)] DEBUG CustomMap  -   row : [1, 2, 3, 4, 5]
2019-06-27 13:40:02.859 [Source: Custom Source -> Map -> Map -> Sink: Unnamed (5/12)] DEBUG CustomMap  -   getArity: 1 
java.lang.ClassCastException: class [Ljava.lang.Integer; cannot be cast to class java.util.List ([Ljava.lang.Integer; and java.util.List are in module java.base of loader 'bootstrap')

注意：

• 尝试解析变量test 也不起作用：java.lang.ClassCastException: class org.apache.flink.types.Row cannot be cast to class java.util.List (org.apache.flink.types.Row is in unnamed module of loader 'app'; java.util.List is in module java.base of loader 'bootstrap')

Answer 1

以下演员表是错误的：

List<Integer> myList = (List<Integer>)myRow.getField(i); // <--- Error here

这是myRow 到List<Integer> 的转换，然后尝试在该List 上调用getField()。

试试：

List<Integer> myList = (List<Integer>)(myRow.getField(i));

也就是说，在myRow 上执行getField()（这将返回Object），然后将Object 转换为List<Integer>。

编辑：我认为问题在于对象是整数数组，而不是列表。请尝试以下操作：

List<Integer> myList = Arrays.asList((Integer[])(myRow.getField(i)));