前面的文章使用Flink批处理完成数据比对（对账）二讨论了使用Table API来处理数据比对的问题，但有些场景下还会有一些比较复杂的业务需求，如输出的时候要将两边的数据合并在一起输出，这个时候用Table API就不太好完成这样的需求了，这就需要借助底层的DataSet API和Process Function。

这篇文章准备利用DataSet API来完成数据比对的需求，至于流数据的实时比对，下一篇文章将介绍。

编写代码：

核心的思想就是用两个流（DataSet其实也是一种特殊的DataStream）中的数据进行处理，Flink中就具备这样的API。

import com.flink.vo.BankVo;
import com.flink.vo.DiffType;
import com.flink.vo.MergeVo;
import com.flink.vo.PayOrgVo;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.CoGroupOperator;
import org.apache.flink.util.Collector;

/**
 * 利用coGroup实现对账需求<br/>
 */
public class BatchJob3 {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<PayOrgVo> payOrgSource = env.fromElements(new PayOrgVo("113", 1), new PayOrgVo("000", 2), new PayOrgVo("115", 33));
        DataSet<BankVo> bankSource = env.fromElements(new BankVo("000", 2), new BankVo("115", 333), new BankVo("114", 4));

        CoGroupOperator<PayOrgVo, BankVo, MergeVo> merge = payOrgSource.coGroup(bankSource)
                .where(PayOrgVo::getOrderNo)//利用关联字段把两边的数据关联起来
                .equalTo(BankVo::getOrderNo)
                .with(new CoGroupFunction<PayOrgVo, BankVo, MergeVo>() {// with方法会将两边orderNo相同的数据放在同一个方法中处理
                    @Override
                    public void coGroup(Iterable<PayOrgVo> first, Iterable<BankVo> second, Collector<MergeVo> out) throws Exception {
                        // 进入到coGroup方法的数据都是orderNo相同的，如果关联不上的即为空
                        PayOrgVo payOrgVo = null;
                        BankVo bankVo = null;
                        DiffType diffType = null;
                        // 以下代码假定数据具有唯一性，即同一个orderNo下仅有一条数据
                        // 如果重复数据，根据实际情况写下面的代码逻辑
                        for (PayOrgVo vo : first) {
                            payOrgVo = vo;
                        }
                        for (BankVo vo : second) {
                            bankVo = vo;
                        }

                        if (bankVo == null) {// 相同orderNo下，支付机构有数据
                            diffType = DiffType.F113;
                        } else if (payOrgVo == null) {// 银行有数据
                            diffType = DiffType.F114;
                        } else if (payOrgVo.getPayment().equals(bankVo.getPayment())) {// 数据完全一致
                            diffType = DiffType.F000;
                        } else {// orderNo相同但payment不同
                            diffType = DiffType.F115;
                        }
                        // 返回数据
                        out.collect(new MergeVo(diffType, payOrgVo, bankVo));
                    }
                });

        merge.print();
    }
}

通过coGroup、where和equalTo很容易讲两个流中orderNo相同的数据关联在一起，coGroup和join不同，join只会关联key相同的数据，形成一个数据集。而coGroup遇到指定key只有一个数据集中有记录的情况时，会将这个Group和空的Group关联。

源码

源码

总结

可以看到，利用Flink将两方数据关联是非常容易的。笔者在实际业务场景中，有些需求不仅需要关联两方数据的，在下发回盘文件的时候，还要关联上其他方数据的情况（如商户数据），这种情况目前想到的办法有：

1. 在CoGroupFunction中利用key去其他地方（文件、数据库、缓存等）获取到第三方的数据，可能会引入其他中间件；
2. 将两方数据关联后，在去join第三方的数据（至于用join还是继续用coGroup，需要读者自己根据业务场景思考）；
3. 扩展Flink的Function，让它能处理三方甚至多方数据，目前刚学，待研究；
4. ……

如果你有更好的想法，欢迎留言，多多指教。
转载请注明出处