美文网首页Spark_Flink_Hadoop架构
使用Flink批处理完成数据比对(对账)二

使用Flink批处理完成数据比对(对账)二

作者: 李不言被占用了 | 来源:发表于2020-04-10 21:48 被阅读0次

    使用Flink批处理完成数据比对(对账)一中,我们只是简单的实现了F000/F113/F114的情况,如果我的需求场景需要实现F115的场景该怎么办呢?

    编写代码

    在上一篇文章的基础上完成代码如下:

    import org.apache.flink.api.java.DataSet;
    import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.api.java.operators.DataSource;
    import org.apache.flink.api.java.operators.MapOperator;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.java.BatchTableEnvironment;
    
    import java.util.List;
    
    /***
     * <strong>对账流程</strong>
     * <ol>
     * <li>两方文件处理如下:</li>
     * <ul>
     * <li>所有唯一性字段(如OrderNO)存放到一个table1</li>
     * <li>所有唯一性字段+比较字段(如OrderNO+OrderMoney)存放到一个table2</li>
     * </ul>
     * <li>比对
     * <ul>
     * <li>两个文件的table1做差集可以得到F113、F114</li>
     * <li>两个文件的table1做交集可以得到F000+F115</li>
     * <li>两个文件的set2做差集可以得到F113+F115</li>
     * <li>F113+F115去除比较字段,只留下关键字段</li>
     * <li>去除F113+F115中的F113,得到F115</li>
     * <li>去除F000+F115中的F115,得到F000</li>
     * </ul>
     * </ol>
     */
    public class BatchJob2 {
    
        public static void main(String[] args) throws Exception {
            // set up the batch execution environment
            final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    
            // Table Environment
            BatchTableEnvironment tableEnvironment = BatchTableEnvironment.getTableEnvironment(env);
    
            /**
             * 构造两个数据集,实际生产从自己需要的source中获取即可
             */
            // 只包含唯一性(用于关联)字段的数据源
            DataSource<String> dataSourceA_unique = env.fromElements("orderId_1_f113", "orderId_2_f000", "orderId_3_f115");
            DataSource<String> dataSourceB_unique = env.fromElements("orderId_2_f000", "orderId_3_f115", "orderId_4_f114");
            // 包含唯一性字段和比较字段
            DataSource<String> dataSourceA_compare = env.fromElements("orderId_1_f113:payment_1", "orderId_2_f000:payment_2", "orderId_3_f115:payment_33");
            DataSource<String> dataSourceB_compare = env.fromElements("orderId_2_f000:payment_2", "orderId_3_f115:payment_333", "orderId_4_f114:payment_4");
    
            // 转换成table
            Table tableA_unique = tableEnvironment.fromDataSet(dataSourceA_unique);
            Table tableB_unique = tableEnvironment.fromDataSet(dataSourceB_unique);
            Table tableA_compare = tableEnvironment.fromDataSet(dataSourceA_compare);
            Table tableB_compare = tableEnvironment.fromDataSet(dataSourceB_compare);
    
            /**
             * 核心对账逻辑
             */
            Table f113_table = tableA_unique.minusAll(tableB_unique);
            Table f114_table = tableB_unique.minusAll(tableA_unique);
            Table f000_f115_table = tableA_unique.intersect(tableB_unique);
    
            Table f113_f115_compare_table = tableA_compare.minusAll(tableB_compare);
            // 拆分,留下唯一性字段
            Table f113_f115_table = convert(tableEnvironment, f113_f115_compare_table);
    
            Table f115_table = f113_f115_table.minusAll(f113_table);
            Table f000_table = f000_f115_table.minusAll(f115_table);
    
            DataSet<String> f000 = tableEnvironment.toDataSet(f000_table, String.class);
            DataSet<String> f113 = tableEnvironment.toDataSet(f113_table, String.class);
            DataSet<String> f114 = tableEnvironment.toDataSet(f114_table, String.class);
            DataSet<String> f115 = tableEnvironment.toDataSet(f115_table, String.class);
    
    
            /**
             * 输出,实际输出到自己需要的sink即可
             */
            List<String> f000_list = f000.collect();
            List<String> f113_list = f113.collect();
            List<String> f114_list = f114.collect();
            List<String> f115_list = f115.collect();
    
            System.out.println("==============================");
            System.out.println("f000 ->" + f000_list);
            System.out.println("==============================");
            System.out.println("f113 ->" + f113_list);
            System.out.println("==============================");
            System.out.println("f114 ->" + f114_list);
            System.out.println("==============================");
            System.out.println("f115 ->" + f115_list);
    
        }
    
        private static Table convert(BatchTableEnvironment tableEnvironment, Table inputTable) {
            DataSet<String> f000_compare_dataset = tableEnvironment.toDataSet(inputTable, String.class);
            MapOperator<String, String> map = f000_compare_dataset.map(e -> {
                return e.split(":")[0];// 留下前半段,关键字段
            });
            return tableEnvironment.fromDataSet(map);
        }
    }
    

    中间的处理逻辑在代码中对注释清楚了。

    源码

    源码

    总结

    需要知道两边都有数据(订单号相同)但存在差异的情况需要处理的步骤多点。
    如果你有更好的想法,欢迎留言,多多指教。
    转载请注明出处

    相关文章

      网友评论

        本文标题:使用Flink批处理完成数据比对(对账)二

        本文链接:https://www.haomeiwen.com/subject/kpxrmhtx.html