[Feature][Core] Design of Dirty Data Collection

[TyrantLucifer]

Search before asking

• I had searched in the feature and found no similar feature requirement.

Description

Design of Dirty Data Collection Functionality for Apache SeaTunnel

Introduction

The dirty data collection function can effectively guide users to discover data quality problems in a timely manner in the data integration framework, ensuring the accuracy and stability of data synchronization.

Functional Requirements

The dirty data collection function needs to support the following features:

1. Collect and record all dirty data content and exception information.
2. Support plug-in extension.

Technical Solution

Interface Design

Add interface DirtyRecordCollector:

/** Base interface for dirty records collector */
public interface DirtyRecordCollector extends Serializable {

    /** Collect dirty record with exception and error message */
    void collect(
            final int subTaskIndex,
            final SeaTunnelRow dirtyRecord,
            final Throwable exception,
            final String errorMessage);

    /** Collect dirty record with exception */
    default void collect(
            final int subTaskIndex, final SeaTunnelRow dirtyRecord, final Throwable exception) {
        collect(subTaskIndex, dirtyRecord, exception, "");
    }
}

Add method getDirtyRecordCollector to SinkWriter.Context:

interface Context extends Serializable {

    /** @return The index of this subtask. */
    int getIndexOfSubtask();

    /** @return metricsContext of this reader. */
    MetricsContext getMetricsContext();

    /**
     * Get dirty record collector
     */
    DirtyRecordCollector getDirtyRecordCollector();
}

Add method getPluginConfig to SeaTunnelPluginLifeCycle:

/**
 * Get the config of plugin
 */
Config getPluginConfig();

Configuration File Example

env {
  job.mode = BATCH
  dirty.collector = {
    type = log
  }
}

source {
  FakeSource {
    row.num = 100
    schema {
      fields {
        name = string
        age = int
      }
    }
  }
}

sink {
  Console {}
}

Technical Implementation

1. Modify seatunnel-config-shade. During the transmission process, SeaTunnelSink will undergo serialization operations, but the underlying interface of typesafe-config does not support serialization, so it is necessary to make Config support serialization to ensure that it is not lost during transmission. See [Feature][Shade][typesafe-config] Make Config can be serialized #4586 for specific implementation details.
2. Modify all SeaTunnelSink to implement the getPluginConfig interface and inject the pluginConfig during the prepare phase.
   Before instantiating the Sink, merge the plugin configuration for all Sinks, and merge the dirty data collector configuration information from the env section into the plugin configuration.
3. Modify DefaultSinkWriterContext and SinkWriterContext to instantiate the DirtyRecordCollector using the configuration and implement the getDirtyRecordCollector method.
4. Modify all SeaTunnelSink to report and update metrics using the dirty data collector when data writing or conversion fails.

Process Design

Usage Scenario

No response

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[Hisoka-X]

so it is necessary to make Config support serialization to ensure that it is not lost during transmission.

Why not use ReadonlyConfig? I think we would remove use config in connector layer in the future.

Before instantiating the Sink, merge the plugin configuration for all Sinks, and merge the dirty data collector configuration information from the env section into the plugin configuration.

I think sink should not care about dirty data collector config. Why not just use env config to create DirtyRecordCollector then put it into context? And Any dirty data collector config should be handle in DirtyRecordCollector .

[TyrantLucifer]

1. Does ReadonlyConfig has the merge ability?
2. The initialize process of SinkWriter.Context happened after sink create, so env config it can not take and we should merge the env config into sink config before sink create.

[hailin0]

Source maybe need collect dirty data

[EricJoy2048]

Source maybe need collect dirty data

+1

[EricJoy2048]

default void collect( final int subTaskIndex, final SeaTunnelRow dirtyRecord, final Throwable exception)

Can we add a CatalogTable type parameter in this method? CatalogTable information can help user Identify which table has the problem.

[liugddx]

Can we add a threshold? Maybe the dirty data has reached the threshold, so the task must be dropped abnormally and an alarm will be issued.

[github-actions]

This issue has been automatically marked as stale because it has not had recent activity for 30 days. It will be closed in next 7 days if no further activity occurs.

[github-actions]

This issue has been automatically marked as stale because it has not had recent activity for 30 days. It will be closed in next 7 days if no further activity occurs.

[github-actions]

This issue has been closed because it has not received response for too long time. You could reopen it if you encountered similar problems in the future.