private void startDataFlow(AuditCycle cycle, int backTimes, Cache<String, StatData> cache) {
        DataQueue dataQueue = new DataQueue(queueSize);
        JdbcSource source = new JdbcSource(dataQueue, buildMysqlSourceConfig(cycle, backTimes));
        source.start();
        dataFlowSources.add(source);

        AuditSink sink;
        if (cache != null) {
            sink = new CacheSink(dataQueue, cache);
        } else {
            SinkConfig sinkConfig = buildMysqlSinkConfig(configuration.get(KEY_MYSQL_SINK_INSERT_DAY_SQL,
                    DEFAULT_MYSQL_SINK_INSERT_DAY_SQL));
            sink = new JdbcSink(dataQueue, sinkConfig);
        }
        sink.start();
        dataFlowSinks.add(sink);
    }
Etl service aggregate the data from the data source and store the aggregated data to the target storage.
inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/service/node/EtlService.java