FLINK:从行字段中提取数组

Question

我正在使用Flink，并且我在地图中使用自定义功能。此自定义函数使用Flink Row类型作为输入，并输出包含我行的每个字段和值的(字符串，对象(的地图。

在基本情况下，此功能很好地工作，但是现在我需要在特定字段上进行一些处理，这是整数数组。在这种情况下，我在将数据中的数据提取到Java Collection对象(列表，数组或其他(时遇到了一些困难。

这是我的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)));