Algo reduce Group函数实现MySQL group_concat(distinct)

标签：

    Algo、reduceGroup、group_concat(distinct)、groupconcat

需求背景：

    MySQL的group_concat([distinct] 要连接的字段 [order by 排序字段 asc/desc ] [separator '分隔符'] ) 可实现将group by产生的同一个分组中的值连接起来，返回一个字符串结果。通过使用distinct可以排除重复值，比如源单数据如图1，经过select suppliername, group_concat( distinct materialname) from kded_purchaseapply group by suppliername;可实现根据供应商分组后将物料名称进行去重连接（如图2）。

图 1 源单数据

图 2 group_concat(distinct)数据

    但是Algo当前版本groupConcat API还不支持使用distinct，未来版本可能支持。目前可在调用GroupbyDataSet的groupConcat API后自己使用DataSet.map()或者GroupbyDataSet.reduceGroup()进一步处理，或者继承CustomAggFunction实现自定义groupby函数。

    本案例演示GroupbyDataSet.reduceGroup()实现group_concat(distinct 要连接的字段)，如有其他类似需求仅供参考。

实现方案：

    数据来源：二开单据采购申请单（kded_purchaseapply）

                    单据头字段 供应商（kded_org）

                    分录（entryentity）字段 物料（kded_materiel）

实现步骤如下：

1. 继承报表模板新建一个报表，添加字段[供应商]（kded_suppliername）、[物料]（kded_materialname)。

2. 继承AbstractReportListDataPlugin实现报表取数插件逻辑，先根据供应商名称group by组并调用groupConcat连接物料名称，groupConcat需要先finish返回dataset，但此时物料名称存在重复值，后面进一步处理。

实现代码：

QFilter filter = new QFilter("billno", QCP.equals, "APPLY-20211125-0001");
DataSet originalDs = QueryServiceHelper.queryDataSet(this.getClass().getName(), "kded_purchaseapply", "kded_supplier.name as kded_suppliername, entryentity.kded_materiel.name as kded_materialname", filter.toArray(), null);
//先根据kded_suppliername分组，调DataSet的groupConcat拼接，此时无法去重，需在调用reduceGroup增加处理
DataSet groupConcatDs = originalDs.groupBy(new String[] {"kded_suppliername"}).groupConcat("kded_materialname").finish();

3. 再次groupBy后调用reduceGroup重写ReduceGroupFunction的reduce方法进行去重。

    (1) 重写getResultRowMeta方法返回DataSet的RowMeta，可用RowMetaFactory.createRowMeta(Object[], DataType[])自己构建，或者用DataSet.getRowMeta()。

    (2) 重写reduce方法对调用DataSet.groupby()的数据行进行数据加工，去除重复值后重新连接返回。

实现代码：

//再次groupBy后调用reduceGroup重写ReduceGroupFunction的reduce方法进行去重
DataSet ds = groupConcatDs.groupBy().reduceGroup(new ReduceGroupFunction() {
private static final long serialVersionUID = 7503741900310230836L;
 
    @Override
    public RowMeta getResultRowMeta() {
        //返回的RowMeta结构与map方法返回Object[]结构保持一致，也可用RowMetaFactory.createRowMeta(Object[], DataType[])自己构建
        //String[] fieldNames = new String[] {"kded_suppliername", "kded_materialname"};
        //DataType[] dataTypes = new DataType[]{DataType.StringType, DataType.StringType};
        //RowMetaFactory.createRowMeta(fieldNames, dataTypes);
        return groupConcatDs.getRowMeta();
    }
    @Override
    public Iterator<Object[]> reduce(Iterator<Row> it) {
        ArrayList<Object[]> arrays = new ArrayList<>();
        while(it.hasNext()) {
            Row row = it.next();
            //去重实现group_concat(distinct)的效果
            String materialName = (String) row.get("kded_materialname");
            String[] split = materialName.split(",");
            List<String> distinctList = Arrays.asList(split).stream().distinct().collect(Collectors.toList());
            String newMaterialName = String.join(",", distinctList);
            arrays.add(new Object[] {row.get("kded_suppliername"), newMaterialName});
        }
        return arrays.iterator();
    }
});

4. 注册报表取数插件。

实现效果：

1. 原始数据。

2.groupConcat(distinct)报表取数。