Cleaning Rules

Tip

To meet data governance needs in different business scenarios, the system provides the "cleaning rules" feature. This article explains how to use this feature to configure and extend the required data cleaning rules independently, supporting diverse data processing scenarios.

1. Cleaning Rule Table Definition

Cleaning rule metadata is stored uniformly in the ATT_CLEAN_RULE table. This table maintains basic information, usage scenarios, strategy identifiers, and related content for cleaning rules.

Table creation statement

-- Create table
CREATE TABLE QDATA_TEST.ATT_CLEAN_RULE (
    ID BIGINT IDENTITY(1,1) NOT NULL,
    NAME VARCHAR2(128) NOT NULL,
    "LEVEL" CHAR(1) DEFAULT '0' NOT NULL,
    DESCRIPTION VARCHAR2(512) NULL,
    VALID_FLAG VARCHAR2(1) DEFAULT '1' NOT NULL,
    DEL_FLAG VARCHAR2(1) DEFAULT '0' NOT NULL,
    CREATE_BY VARCHAR2(32) NULL,
    CREATOR_ID BIGINT NULL,
    CREATE_TIME DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
    UPDATE_BY VARCHAR2(32) NULL,
    UPDATER_ID BIGINT NULL,
    UPDATE_TIME DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
    REMARK VARCHAR2(512) NULL,
    EXAMPLE VARCHAR2(512) NULL,
    "TYPE" VARCHAR2(10) NULL,
    USE_CASE VARCHAR2(512) NULL,
    STRATEGY_KEY VARCHAR2(32) NULL,
    CODE VARCHAR2(32) NULL,
    CAT_CODE VARCHAR2(128) NULL,
    CONSTRAINT PK_ATT_CLEAN_RULE PRIMARY KEY (ID)
);
-- Table comment:
COMMENT ON TABLE QDATA_TEST.ATT_CLEAN_RULE IS 'Cleaning rule table';

Field description

Field Name	Type	Constraint/Default	Description
ID	BIGINT	Primary key, auto-increment	Rule ID
NAME	VARCHAR2(128)	NOT NULL	Rule name
LEVEL	CHAR(1)	Default '0', NOT NULL	Rule level; 1: field level, 2: table level
DESCRIPTION	VARCHAR2(512)		Rule description
VALID_FLAG	VARCHAR2(1)	Default '1', NOT NULL	Whether valid; 0: invalid, 1: valid
DEL_FLAG	VARCHAR2(1)	Default '0', NOT NULL	Deletion flag; 1: deleted, 0: not deleted
CREATE_BY	VARCHAR2(32)		Creator
CREATOR_ID	BIGINT		Creator ID
CREATE_TIME	DATETIME	Default CURRENT_TIMESTAMP	Creation time
UPDATE_BY	VARCHAR2(32)		Updater
UPDATER_ID	BIGINT		Updater ID
UPDATE_TIME	DATETIME	Default CURRENT_TIMESTAMP	Update time
REMARK	VARCHAR2(512)		Remarks
EXAMPLE	VARCHAR2(512)		Example
TYPE	VARCHAR2(10)		Cleaning rule category ID
USE_CASE	VARCHAR2(512)		Usage scenario
STRATEGY_KEY	VARCHAR2(32)		Strategy identifier, used for rule recognition during cleaning component execution
CODE	VARCHAR2(32)		Rule code
CAT_CODE	VARCHAR2(128)		Category code

---

Key field: STRATEGY_KEY
This field is used for rule recognition and dispatch during cleaning component execution. It is required when extending new rules.

2. Frontend Parameterized Integration

After basic information and identifiers, including STRATEGY_KEY, are configured in the cleaning rule table, the cleaning rule appears in the ETL cleaning component list.

At this point, clicking it still has no effect, so frontend extension is required:

Development notes

1. Frontend page development

   • Create the corresponding configuration page for the new cleaning rule.
   • The page is used to enter and validate the parameters required by the rule.

2. Parameter encapsulation

   • Encapsulate frontend configuration parameters into the ETL component's taskParams.
   • The parameter structure must remain consistent with the parsing logic of the backend parse2 method.

3. Pass to backend

   • When the frontend saves, parameters are written into the ETL task definition together with node information.
   • During later task execution, parameters are passed to the execution component through taskParams.

---

Note:

• If a new cleaning rule has no extra parameters, it can directly inherit the default structure;
• If extension is needed, define new parameter fields in taskParams and update the parsing logic accordingly.

3. ETL Parameter Encapsulation and Cleaning Component Parameter Binding

After the frontend saves ETL component information, the system enters the parameter encapsulation stage.
The related logic is located under the qdata-module-dpp/qdata-module-dpp-biz module:

tech.qiantong.qdata.module.dpp.utils.TaskConverter#buildEtlTaskParams

Method description

/**
 * Build ETL parameter data
 */
public static Map<String, Object> buildEtlTaskParams(
        String taskDefinitionList,
        Map<String, DppEtlNodeSaveReqVO> nodeMap,
        Map<String, Object> taskInfo,
        List<DsResource> resourceList) {

    Map<String, Object> result = new HashMap<>();
    List<Map<String, Object>> transitionList = new ArrayList<>();
    List<DppEtlNodeSaveReqVO> nodeList = JSON.parseArray(taskDefinitionList, DppEtlNodeSaveReqVO.class);

    for (DppEtlNodeSaveReqVO dppEtlNodeSaveReqVO : nodeList) {
        Integer version = 1;
        if (nodeMap.containsKey(dppEtlNodeSaveReqVO.getCode())) {
            version = nodeMap.get(dppEtlNodeSaveReqVO.getCode()).getVersion();
        }

        // Component type
        String componentType = dppEtlNodeSaveReqVO.getComponentType();
        TaskComponentTypeEnum taskComponentTypeEnum = TaskComponentTypeEnum.findEnumByType(componentType);

        Map<String, Object> data = ComponentFactory.getComponentItem(componentType)
                .parse2(dppEtlNodeSaveReqVO.getCode(), version, taskComponentTypeEnum,
                        dppEtlNodeSaveReqVO.getTaskParams(), resourceUrl, resourceList);

        data.put("nodeName", dppEtlNodeSaveReqVO.getName());
        data.put("projectCode", taskInfo.get("projectCode"));

        switch (taskComponentTypeEnum) {
            case DB_READER:
            case EXCEL_READER:
            case CSV_READER:
                result.put("reader", data);
                break;
            case SPARK_CLEAN:
            case SORT_RECORD:
            case FIELD_DERIVATION:
                transitionList.add(data);
                break;
            case DB_WRITER:
                result.put("writer", data);
                break;
        }
    }

    // Configure config
    Map<String, Object> config = new HashMap<>();
    config.put("taskInfo", taskInfo);
    config.put("rabbitmq", rabbitmqConfig);
    config.put("resourceUrl", resourceUrl);

    result.put("transition", transitionList);
    result.put("config", config);

    return result;
}

Key points

• Component classification handling

  • reader: DB / Excel / CSV readers
  • transition: intermediate processing chain, such as cleaning, sorting, and field derivation
  • writer: DB writer

• Cleaning component type

  • SPARK_CLEAN is the type of the cleaning component. At this stage, it is added to transitionList.

• Parameter passing

  • Parameters encapsulated by the cleaning rule page enter taskParams.
  • In the parse2 method, these parameters are parsed and written into transition.
  • Common metadata, such as nodeName, projectCode, and version, is completed uniformly by buildEtlTaskParams.

---

Note

• When developing a new cleaning rule, there is no need to modify the buildEtlTaskParams method.
• Just ensure that the taskParams parameter structure passed from the frontend is consistent with the parse2 method parsing logic.

4. Cleaning Component Parameter Encapsulation and Extension

Cleaning component parameters are encapsulated in buildEtlTaskParams by calling the parse2 method:

Map<String, Object> data = ComponentFactory.getComponentItem(componentType)
        .parse2(dppEtlNodeSaveReqVO.getCode(),
                version,
                taskComponentTypeEnum,
                dppEtlNodeSaveReqVO.getTaskParams(),
                resourceUrl,
                resourceList);

Method example:

@Override
public Map<String, Object> parse2(String nodeCode,
Integer nodeVersion,
TaskComponentTypeEnum componentType,
Map<String, Object> taskParams,
String resourceUrl,
List<DsResource> resourceList) {
// Reader configuration
Map<String, Object> reader = new HashMap<>();
reader.put("nodeCode", nodeCode);
reader.put("nodeVersion", nodeVersion);
reader.put("componentType", componentType.getCode());

    // Parameters
    Map<String, Object> parameter = new HashMap<>();
    Map<String, Object> mainArgs = (Map<String, Object>) taskParams.get("mainArgs");
    parameter.put("cleanRuleList", mainArgs.get("cleanRuleList"));
    parameter.put("tableFields", taskParams.get("tableFields"));
    parameter.put("where", taskParams.get("where"));
    reader.put("parameter", parameter);

    return reader;
}

Key points

• Parameter encapsulation

  • All cleaning component parameters are stored uniformly in taskParams.
  • In the parse2 method, these parameters are parsed and transferred into the parameter field.

• Default structure

  • cleanRuleList: cleaning rule collection
  • tableFields: current table field information
  • where: conditional filter expression

• Extension description

  • In general, there is no need to modify the default logic of parse2.
  • If extension is needed, add new parameter fields to taskParams and store them in parse2.

---

Note

• The parameter model of the frontend page must remain consistent with the parsing logic of the parse2 method.
• When adding new parameter fields, maintain backward compatibility to avoid affecting existing rule execution.

5. Execution-side Implementation (Spark Cleaning Execution)

Cleaning rule execution is located in the qdata-etl module.
Class: tech.qiantong.qdata.spark.etl.transition.CleanTransition

Key flow (excerpt)

JSONObject parameter = transition.getJSONObject("parameter");
// 1) Get parameters
List<Map<String, Object>> tableFieldList = (List<Map<String, Object>>) parameter.get("tableFields");
if (tableFieldList == null || tableFieldList.isEmpty()) {
    return transitionOld(dataset, transition, logPath);
}
// 2) Global where
String where = parameter.getString("where");
if (StringUtils.isNotEmpty(where)) {
    dataset = safeFilter(dataset, where, logPath);
}

// 3) Process rules one by one
for (Map<String, Object> rule : tableFieldList) {
    String ruleCode = MapUtils.getString(rule, "ruleCode");
    String ruleType = MapUtils.getString(rule, "ruleType");
    JSONObject ruleConfig = JSONObject.parseObject((String) rule.get("ruleConfig"));
    String whereClause = MapUtils.getString(rule, "whereClause");

    if (StringUtils.isNotEmpty(whereClause)) {
        dataset = safeFilter(dataset, whereClause, logPath);
    }

    // Field existence validation
    if (!checkColumnsExist(dataset, ruleConfig)) {
        LogUtils.writeLog(logPath, String.format("Skip rule %s (field does not exist)", ruleCode));
        continue;
    }

    // 4) Dispatch execution
    switch (ruleType) {
        case "WITHIN_BOUNDARY":        // Numeric boundary adjustment
            dataset = applyNumericBoundary(dataset, ruleConfig);
            break;
        case "REMOVE_EMPTY_COMBINATION": // Delete when combined fields are empty
            dataset = applyDeleteIfAllNull(dataset, ruleConfig);
            break;
        case "ADD_PREFIX_SUFFIX":      // Standardize field prefix/suffix
            dataset = applyPrefixSuffix(dataset, ruleConfig);
            break;
        case "MENU_CUSTOM":            // Enumeration value mapping standardization
            dataset = normalizeEnumMapping(dataset, ruleConfig);
            break;
        case "KEEP_LATEST_OR_FIRST":   // Deduplicate by combined fields, keeping latest or first record
            dataset = deduplicateByFieldsKeepFirst(dataset, ruleConfig);
            break;
        default:
            LogUtils.writeLog(logPath, "Unknown rule: " + ruleCode);
    }
}

Parameter sources

• parameter.tableFields

  • Contains the cleaning rule list
  • Each rule contains: ruleCode / ruleType / ruleConfig / whereClause

• parameter.where

  • Global filter condition
  • Executed before rule-level whereClause

Extension guide

1. Rule registration

   • Add a new record to the cleaning rule table.
   • Set a unique STRATEGY_KEY, which must correspond to the frontend ruleType.

2. Frontend modeling

   • Add the parameter model for the rule to the cleaning component configuration page.
   • Write parameters into taskParams.mainArgs.cleanRuleList[*].ruleConfig.

3. Execution dispatch

   • Add a new branch in switch (ruleType).
   • The branch name is recommended to remain consistent with STRATEGY_KEY.
   • Call the corresponding implementation method.

4. Rule implementation

   • Add a concrete implementation method in the CleanTransition class, such as applyYourRuleName(dataset, ruleConfig).
   • The method must complete:
     • Field existence validation
     • Data transformation logic
     • Log recording

Notes

• Rule execution order

  • Rules are executed sequentially in the order of the tableFields array.
  • If dependencies exist between rules, define the order clearly on the frontend or during save.

• Field validation

  • Each rule must pass checkColumnsExist validation before execution.
  • For dynamic columns, define a missing-column handling strategy, such as skip, default value, or error.

• Conditional filtering

  • Both global where and rule-level whereClause are executed through safeFilter.
  • Recommended execution order: global filter first, then local filter.
  • Use unified log output to avoid injection risks.

• Extension standards

  • New rules must ensure that STRATEGY_KEY corresponds to ruleType and is unique.
  • Keep the parameter structure consistent with the frontend taskParams definition to avoid parsing exceptions.

6. Notes Overview

When extending and implementing cleaning rules, pay special attention to the following:

• Parameter extension

  • Put new parameters uniformly into ruleConfig to avoid modifying the common structure.

• Database compatibility

  • If cross-database execution is involved, refer to dialect adaptation in the data connection module to ensure type and syntax compatibility.

• Unique identifier

  • STRATEGY_KEY must be globally unique and is used for rule recognition during cleaning component execution.
  • When changing it, handle backward compatibility and smooth migration of historical rules.

• Frontend and backend consistency

  • The frontend parameter model must remain consistent with the backend parse2 parsing logic.
  • Parameter validation should be handled on both frontend and backend to prevent "dirty parameters" from entering the execution side.

• Execution order

  • Rules in transition are executed sequentially in array order.
  • If rules have dependencies, adjust the order during save.

• Conditional filtering

  • Global where and rule-level whereClause can be combined. The execution order is global first, local second.

• Field existence

  • Field validation is recommended before execution.
  • Dynamic columns require a missing-value handling strategy, such as skip, default value, or error.

• Null and type handling

  • Define a clear strategy for null values, such as ignore or default value.
  • Numeric and date fields should use unified cast to avoid implicit conversion.

• Performance optimization

  • Execute filter conditions first to reduce subsequent data volume.
  • Use operations that trigger shuffle cautiously, and add caching when necessary.

• Logs and observability

  • Rule execution should record start, end, duration, and affected row count.
  • Exception logs should include ruleCode and node name for easier troubleshooting.

• Extension standards

  • New rules should provide unit or integration test examples.
  • Rule documentation should be updated accordingly so development and operations teams can reference it.