前面的文章使用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将两方数据关联是非常容易的。笔者在实际业务场景中，有些需求不仅需要关联两方数据的，在下发回盘文件的时候，还要关联上其他方数据的情况（如商户数据），这种情况目前想到的办法有：
- 在
CoGroupFunction中利用key去其他地方（文件、数据库、缓存等）获取到第三方的数据，可能会引入其他中间件； - 将两方数据关联后，在去join第三方的数据（至于用join还是继续用coGroup，需要读者自己根据业务场景思考）；
- 扩展Flink的Function，让它能处理三方甚至多方数据，目前刚学，待研究；
- ……
如果你有更好的想法，欢迎留言，多多指教。
转载请注明出处
