1
2
3
4
5
6
7
8
9
10
11
12
13
/**
 * {
 *     1:[2,3,4],
 *     2:[3,4,5]
 * }
 * ↓
 * {
 *     2:[1],
 *     3:[1,2],
 *     4:[1,2],
 *     5:[2]
 * }
 */

flink由上转成下面的数据格式

  1. 打散

    1
    2
    3
    4
    5
    6
    2:1,
    3:1,
    4:1,
    3:2,
    4:2,
    5:2

    2.再根据key分组并且reduce

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    DataSet<Tuple2<String, String>> test = ...
    test.flatMap(new FlatMapFunction<Tuple2<String, String>, Tuple2<String, String>>() {
    	@Override
    	public void flatMap(Tuple2<String, String> value, Collector<Tuple2<String, String>> out) throws Exception {
    		Tuple2<String, String> tuple2 = new Tuple2<>();
    		String key = value.f0;
    		String values = value.f1;
    		String[] splitCmId = values.split(",");
    		for (String cmId : splitCmId) {
    			tuple2.setField(cmId, 0);
    			tuple2.setField(key, 1);
    			out.collect(tuple2);
    		}
    	}
    }).
    groupBy(0).
    reduce(new ReduceFunction<Tuple2<String, String>>() {
    	@Override
    	public Tuple2<String, String> reduce(Tuple2<String, String> value1, Tuple2<String, String> value2) throws Exception {
    		Tuple2<String, String> tuple2 = new Tuple2<>();
    		tuple2.setField(value1.f0, 0);
    		String f1 = value1.f1 + "," + value2.f1;
    		// 这里可能重复,去重
    		Set<String> set = new HashSet<>(Arrays.asList(f1.split(",")));
            tuple2.setField(String.join(",", set), 1);
    		return tuple2;
    	}
    });