Flink·凯比·波霍

Question

我先做一个keyBy，然后做一个聚合，但是Flink没有正确地对数据进行分组（相反，每个事件都属于自己的分组by）。

示例：

Class Purchase {
    String product;
    Integer quantity;
}

Class Filter {
    String product;

    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + ((product == bull) ? 0 : displayName.hashCode());
}

Class FilteredPurchase {
    Filter filter;
    Purchase purchase;
}

DataStream<FilteredPurchase> =
    ...
    .keyBy(“filter”) //This works
    .keyBy(x -> x.getFilter()). // This doesn’t 
    .sum(“trade.quantity”);

如果我们考虑如下所示的流的情况：

[
    {“filter”: {“product”: null}, “purchase”: {“product”: “apple”, “quantity”: 10},
    {“filter”: {“product”: null}, “purchase”: {“product”: “apple”, “quantity”: 10},
    {“filter”: {“product”: “apple”}, “purchase”: {“product”: “apple”, “quantity”: 10},
    {“filter”: {“product”: “apple”}, “purchase”: {“product”: “apple”, “quantity”: 10},
]

Answer 1

第一个问题是为什么不使用product（String）作为键，因为这是filter类中的全部内容。所以

.keyby(x->x.getproduct())

但无论如何，我认为您的键类(filter)必须实现equals()方法。