Algo map函数实现MySQL的group_concat(distinct)

标签：

    Algo、map、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函数。

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

实现方案：

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

                    单据头字段 供应商（kded_org）

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

实现步骤如下：

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

2. 继承MapFunction实现MyMapFunction，重写getResultRowMeta和map方法。

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

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

实现代码：

class MyMapFunction extends MapFunction {
    private static final long serialVersionUID = 196453238344939218L;
    @Override
    public RowMeta getResultRowMeta() {
        //返回的RowMeta结构与map方法返回Object[]结构保持一致
        String[] fieldNames = new String[] {"kded_suppliername", "kded_materialname"};
        DataType[] dataTypes = new DataType[]{DataType.StringType, DataType.StringType};
        return RowMetaFactory.createRowMeta(fieldNames, dataTypes);
    }
    @Override
    public Object[] map(Row row) {
        String orgname = (String) row.get("kded_materialname");
        String[] split = orgname.split(",");
        List<String> distinctList = Arrays.asList(split).stream().distinct().collect(Collectors.toList());
        String newOrgName = String.join(",", distinctList);
        return new Object[] {row.get("kded_suppliername"), newOrgName};
    }
}

3. 继承AbstractReportListDataPlugin实现报表取数插件逻辑。

实现代码：

@Override
public DataSet query(ReportQueryParam queryParam, Object object) throws Throwable {
    QFilter filter = new QFilter("billno", QCP.equals, "APPLY-20211125-0001");
    //获取原始数据集DataSet
    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);
    //根据供应商名称分组后连接物料名称
    DataSet groupDs = originalDs.groupBy(new String[] {"kded_suppliername"}).groupConcat("kded_materialname").finish();
    //调用Map函数去重实现groupConcat(distinct)
    return groupDs.map(new MyMapFunction());
}

4. 注册报表取数插件。

实现效果：

1. 原始数据。

2.groupConcat(distinct)报表取数。