Flink: извлечь массив из поля строк - PullRequest
Новая
       19

Flink: извлечь массив из поля строк

0 голосов
/ 27 июня 2019

Я использую Flink, и я использую пользовательскую функцию на карте.Эта пользовательская функция использует тип строки Flink в качестве ввода и вывода карты (String, Object), которая содержит каждое поле и значения моей строки.

В базовом случае эта функция работает хорошо, но теперь мне нужно сделатьнекоторая обработка по определенному полю, которое является массивом целых чисел.В этом случае у меня возникают некоторые проблемы при извлечении данных из моей строки в объект 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')

1 Ответ

1 голос
/ 27 июня 2019

Следующее приведение неверно: 

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

Это приведение myRow к List<Integer>, затем попытка вызвать getField() для этого List.

Попробуйте: 

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

, то есть выполните getField() на myRow (который вернет Object), затем приведите это Object к List<Integer>.

РЕДАКТИРОВАТЬ: Я думаю, что проблема заключается в том, что объект представляет собой массив целых чисел, а не список.Попробуйте следующее: 

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