В примере кода ниже я пытаюсь получить поток записей сотрудников {страна, работодатель, имя, зарплата, возраст} и сбросить наибольшего платного работника в каждой стране. К сожалению, множественные ключи от не работают. < /P>
Только ключ (работодатель) < /code> отражается, поэтому я не получаю правильный результат.
Что мне не хватает? < /P>
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream streamEmployee = env.addSource(
new FlinkKafkaConsumer010("flink-demo", new JSONDeserializationSchema(), properties))
.map(new MapFunction() {
private static final long serialVersionUID = 6111226274068863916L;
@Override
public Employee map(ObjectNode value) throws Exception {
final Gson gson = new GsonBuilder().create();
Employee uMsg = gson.fromJson(value.toString(), Employee.class);
return uMsg;
}
});
KeyedStream employeesKeyedByCountryndEmployer = streamEmployee
.keyBy(new KeySelector() {
private static final long serialVersionUID = -6867736771747690202L;
@Override
public String getKey(Employee value) throws Exception {
// TODO Auto-generated method stub
return value.getCountry();
}
}).keyBy(new KeySelector() {
private static final long serialVersionUID = -6867736771747690202L;
@Override
public String getKey(Employee value) throws Exception {
// TODO Auto-generated method stub
return value.getEmployer();
}
});
// This should display employees highly paid in a given country , for a
// given employer
DataStream uHighlyPaidEmployee = employeesKeyedByCountryndEmployer.timeWindow(Time.seconds(5))
.maxBy("salary");
// Assume toString() is overridden , so print works well.
uHighlyPaidEmployee.print();
env.execute("Employee-employer log processor");
Подробнее здесь: https://stackoverflow.com/questions/462 ... iple-keyby
