github-actions[bot] commented on issue #18626:
URL:
https://github.com/apache/dolphinscheduler/issues/18626#issuecomment-5615570489
# DolphinScheduler Lake warehouse asset version event-driven scheduling
design document
> Status: Draft/RFC
> Author: Dolphin Agent
> Associated modules: dolphinscheduler-dao, dolphinscheduler-service,
dolphinscheduler-master, dolphinscheduler-api, dolphinscheduler-task-plugin
## 1. Background and goals
### 1.1 Background
The modern CDC lake warehouse architecture (based on Paimon / Iceberg / Hudi
and other table formats) is promoting the evolution of data warehouse modeling
and data platform scheduling paradigms:
- Data warehouse modeling is upgraded from a single "table" to a three-layer
semantics of "latest-state table (latest-state) + event fact table (event fact
table) + change history table (changelog)", relying on the Snapshot, Manifest,
Primary Key Table, Changelog, Time Travel/Tag, Schema Evolution and other
capabilities provided by the table format.
- The scheduling paradigm is upgraded from "time DAG" to "data asset status
scheduling": that is, the triggering basis of scheduling is no longer just the
cron time point, but "whether the data asset has been advanced to a certain
available version." Typical scenarios include:
- Trigger downstream processing after snapshot generation
- Trigger the window to calculate the output after the watermark arrives
- Indicators are only allowed to be released after passing the quality
check
- After the tag is solidified, it triggers the issuance of financial
statements.
- Trigger downstream model upgrade after schema change approval is passed
- Trigger recalculation and reconciliation after backfill is completed
### 1.2 Goals
Without destroying DolphinScheduler's existing scheduling semantics (cron
timing, manual running, complement, failure recovery, DEPENDENT tasks, normal
DAG execution), the "data asset version promotion event-driven scheduling"
capability is added to achieve:
```text
Lake warehouse table snapshot/instant promotion
->The event is sensed and persisted as an AssetEvent
-> Update the current state of the asset AssetState (latest snapshot /
watermark / quality / schema)
-> Matching evaluation based on the asset dependency conditions of the
workflow/task statement
-> After the dependencies are satisfied, trigger/release the
DolphinScheduler workflow or task instance through the Command mechanism
-> Execution, observation, failure retry and compensation
```
The design follows the following non-negotiable principles:
1. It cannot be "triggered when the event arrives", it must be "triggered
only when the asset status meets the conditions".
2. The same version/same dependency combination can only be triggered once
(idempotent).
3. Events may be repeated, out of sequence, delayed or lost. Asset status is
the basis for scheduling, and events are only input evidence.
4. Prioritize the reuse of existing Command / CommandType / DEPENDENT task /
task instance state machine and other infrastructure, and do not create a new
independent scheduling kernel.
5. In the first phase, only a small range of verifiable closed loops (Paimon
snapshot -> AssetEvent -> AssetState -> Dependency matching -> Command trigger)
will be implemented, and capabilities such as Iceberg/Hudi and multi-asset
AND/OR combinations will be expanded later.
### 1.3 Applicable boundaries and problem definition
This design is oriented to the **new generation data warehouse
architecture** (based on Paimon/Iceberg/Hudi and other lake warehouse table
formats). Its core features are:
- **The entire DAG link is based on snapshot version driver**: from ODS
(data into the lake) to DWD (dimensional processing) to DWS (theme aggregation)
to ADS (external services), each layer is the result of the advancement of the
data asset version of the previous layer, eliminating the mismatch between time
and data availability.
- **Does not rely on cron and time expressions**: The triggering of any task
in the DAG is based on its dependent assets (snapshot/watermark/quality/tag),
rather than "a fixed moment".
- **Downstream is triggered only when multiple upstream assets are ready at
the same time**: For example, a task in DWS may depend on the snapshots of
multiple tables in DWD being advanced. These dependencies are expressed through
snapshot version combinations, rather than time windows.
- **Retain orchestration capabilities such as cross-stream dependencies**:
If a task in the DAG declares cross-stream dependencies (depending on tasks in
other workflows/projects), these dependencies should also be upgraded to be
based on the snapshot model to ensure the data-driven nature of the overall
scheduling.
**Applicable scenarios**: Any data processing workflow based on the lake
warehouse table carrying the full link of ODS/DWD/DWS/ADS.
**Not applicable scenarios**: Traditional data warehouses that are still
using "fixed time batch processing" (even if they have been moved to the cloud).
## 2. Current situation analysis: DolphinScheduler’s existing
event/dependency triggering infrastructure
Through actual inspection of the code base (version: current master branch),
the existing reusability capabilities are as follows:
### 2.1 CommandType/Command(`dolphinscheduler-dao`,
`dolphinscheduler-common`)
- `org.apache.dolphinscheduler.common.enums.CommandType` (file:
`dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/CommandType.java`)
is defined `START_PROCESS`, `SCHEDULER`, `COMPLEMENT_DATA`,
`RECOVER_TOLERANCE_FAULT_PROCESS` and other trigger source types (`// todo:
rename to WorkflowTriggerType` has been marked in the comments, indicating that
the official has regarded Command as an abstraction of "trigger type", not just
a "start action").
- `org.apache.dolphinscheduler.dao.entity.Command`
(`dolphinscheduler-dao/.../entity/Command.java`) is a drop-in entity,
corresponding to the table `t_ds_command`, core fields: `commandType`,
`workflowDefinitionCode`, `workflowDefinitionVersion`, `commandParam` (JSON
format startup parameters), `workflowInstancePriority`.
- `org.apache.dolphinscheduler.service.command.CommandService`
(`dolphinscheduler-service/.../command/CommandService.java`) exposes `int
createCommand(Command command)`, which is the unified entrance for all current
"initiating a workflow instance run". The `CommandService`/scanner on the
Master side will consume the `t_ds_command` table to generate WorkflowInstance.
**Conclusion**: We do not need to create a new parallel "event-triggered
workflow" execution channel. Add a new `CommandType.ASSET_EVENT_TRIGGER` (or
reuse the trigger source marked in `START_PROCESS` + `commandParam`), and write
`t_ds_command` through the existing `CommandService#createCommand` to
completely reuse the existing Command consumption, WorkflowInstance creation,
and DAG execution links on the Master side. This is the least risky integration
point.
### 2.2 DEPENDENT task (task-level dependent waiting)
- `DependentType`, `DependentRelation`
(`dolphinscheduler-task-plugin/dolphinscheduler-task-api/.../enums/DependentType.java`,
`DependentRelation.java`): Currently DEPENDENT only supports dependencies on
"whether another workflow/task is successful within a certain cycle"
(`DependentItem` contains
`projectCode`/`definitionCode`/`depTaskCode`/`cycle`/`dateValue`/`dependResult`),
**aimed at "task instance execution results", not "external data asset
status"**.
- `DependentParameters` (same directory
`parameters/DependentParameters.java`) supports `DependentRelation` (AND/OR) to
combine multiple `DependentTaskModel`. The structure of this "multi-dependency
AND/OR combination evaluation" can directly draw on the combined expression for
asset dependency.
- `DependentLogicTask` / `DependentTaskTracker`
(`dolphinscheduler-master/.../executor/plugin/dependent/`): After the task
instance enters the execution state, `DependentTaskTracker` polls whether the
dependency is reached, `getDependentTaskStatus()` returns `TaskExecutionStatus`
essentially means "the task instance occupies an execution slot and polls and
waits internally", and is not event callback driven.
**Conclusion**: The "multi-dependency AND/OR combination expression" of the
DEPENDENT task is worth reusing its design (`DependentRelation`), but its
"dependent object" is the task execution result rather than the external data
asset version, so its `DependentItem`/`DependentTaskTracker` implementation
cannot be directly reused. A new asset-oriented dependency model and evaluator
need to be added; the execution form can refer to it as the new Task type
`ASSET_SENSOR` (strategy A, see below), reuses its operating mode and state
machine (`onTaskRunning`/`onTaskPaused`/`onTaskKilled` and other existing life
cycle hooks in `AbstractLogicTask`) of "occupying task instance slot + polling
determination + entering downstream after success".
### 2.3 No capabilities (need to be added)
- There is no Airflow Dataset style "Asset" first-class citizen model, no
`t_ds_asset` class table.
- There is no unified access layer for external events
(messages/callbacks/polling); the `dolphinscheduler-extract` module is
currently mainly an internal RPC contract between Master/Worker/API
(`extract-master`, `extract-worker`, `extract-alert`, etc.), and there is no
event access interface for external lake warehouse systems.
- There is no "event deduplication + idempotent trigger history" table, and
currently `t_ds_command` has no unique constraints to prevent repeated
insertion of Commands for the same (workflow, trigger condition) (the business
layer is controlled through `scheduler`/manual trigger points, and the
"external event driven" scenario is not covered).
- There are no fields and state machines with lake warehouse semantics such
as watermark / snapshot / quality / schema.
In summary, **overall strategy**: add three new logical modules
`event-source`, `asset-state`, `dependency-resolver` and supporting DAO tables,
connect to the existing `CommandService.createCommand` (Strategy C, directly
create workflow instances) on the trigger side, and add a new `ASSET_SENSOR`
task type (Strategy A, task-level waiting, reuse DAG embedded waiting
semantics), the two strategies are implemented in stages, which is not correct
The Master schedules intrusive modifications to the kernel and task state
machines.
## 3. Overall architecture design
### 3.1 Overall link
```text
|
│ Lakehouse │ │ Event Ingestion│ │ Asset State │ │ Dependency Resolver │ │
Scheduler │
│ (Paimon/Iceberg│──▶│ (Poll Scanner /│──▶│ Store │──▶│ (Match
AssetDependency│──▶│ Adapter │
│ /Hudi commit) │ │ Report API) │ │ (t_ds_asset_ │ │ -> READY/BLOCKED) │ │
(CommandService│
└───────────────┘ └────────────────┘ │ state) │
└───────────────────────────┘ │ / ASSET_SENSOR│
└───────────────┘ │ Task) │
└───────┬────────┘
▼
Master normal Command/DAG execution link
```
### 3.2 Asset (Asset) and Asset Event (AssetEvent) domain model
```java
// Unique asset identifier: catalog.database.table[/partition]
public class AssetIdentifier {
private String assetKey; // Normalized unique key, such as
paimon://catalog/db/table/dt=2024-05-20
private String storageFormat; // PAIMON / ICEBERG / HUDI
private String catalogName;
private String databaseName;
private String tableName;
private String partitionExpr; // Can be null, supports partition-level
assets
}
// The original event, the source can be polling scan or external active
reporting, is only used as "input evidence" and cannot directly trigger
scheduling
public class AssetEvent {
private String eventId; // One of the idempotent keys, which can be
calculated repeatedly in the source system (see 4.2)
private String assetKey;
private String eventType; // SNAPSHOT_COMMITTED / WATERMARK_ADVANCED /
TAG_CREATED /
// QUALITY_CHECK_PASSED /
QUALITY_CHECK_FAILED /
// SCHEMA_CHANGE_APPROVED /
BACKFILL_COMPLETED / CDC_LAG_RECOVERED
private Long snapshotId; // Paimon snapshot-id / Iceberg snapshot-id /
Hudi instant time mapping value
private Long schemaId;
private Long watermark; // epoch millis
private String commitKind; // APPEND / COMPACT / OVERWRITE / ...
private Long commitTime;
private Long deltaRecordCount;
private Boolean schemaChanged;
private String payload; // Original JSON, convenient for
troubleshooting, does not participate in judgment logic
private Long receiveTime;
}
// Current status of assets: the only source of truth for scheduling
decisions (Source of Truth)
public class AssetState {
private String assetKey;
private Long latestSnapshotId;
private Long latestSchemaId;
private Long latestWatermark;
private String latestTag;
private String qualityStatus; // UNKNOWN / PASSED / FAILED
private String schemaStatus; // UNKNOWN / COMPATIBLE / PENDING_APPROVAL
/ REJECTED
private String backfillStatus; // NONE / RUNNING / COMPLETED
private Long updateTime;
private Long version; // Optimistic locking to prevent concurrent
coverage (see 4.4)
}
```
### 3.3 Design of event acquisition method (Event Ingestion)
The two methods are parallel and complementary to each other, and the MVP
stage gives priority to implementation method 1:
**Method 1: Incremental polling scan (Polling Scanner, priority
implementation)**
- Deploy a lightweight `AssetEventScanner` independently (can be used as a
background task within DolphinScheduler, or an independent process/API
scheduled task), and execute it regularly on the registered Paimon table:
```sql
SELECT snapshot_id, schema_id, commit_user, commit_identifier,
commit_kind, commit_time,
watermark, total_record_count, delta_record_count,
changelog_record_count
FROM catalog_name.database_name.`table_name$snapshots`
WHERE snapshot_id > ?
ORDER BY snapshot_id;
```
where `?` comes from the `last_scanned_snapshot_id` recorded in
`t_ds_asset_event_consumer_offset`.
- Advantages: It does not depend on the writer's modification, and can
directly access the existing Paimon/Iceberg table; it has the natural
"compensation" ability (it is a fully replayable pull model).
- Disadvantages: There is a delay caused by the scanning interval (typically
10s~1min).
**Method 2: The writer actively reports (Push, used in low-latency
scenarios, access in subsequent stages)**
- After the Flink/Spark write job is successfully committed, it calls the
new REST interface `POST /dolphinscheduler/asset-events` to proactively report
events (see 4.6 API design).
- Advantages: low latency (seconds).
- Disadvantages: The business code of the writing party needs to be
modified; and it must coexist with polling scanning to ensure that false
negatives cannot be discovered.
**Conclusion (trade-offs)**: Only method one (Polling Scanner) is
implemented in the MVP stage because it has zero intrusion into the existing
Flink/Paimon production link, can be independently verified, and has natural
compensation properties. Method 2 is a second-phase enhancement. Both share the
same "event deduplication -> status update -> dependency evaluation" processing
link. Only the event source (`sourceType=POLL` / `sourceType=PUSH`) is
different. Through the unique constraint of `t_ds_asset_event`, the dual paths
are naturally deduplicated, and the dual paths will not be triggered repeatedly.
### 3.4 Database table design (DDL draft)
Add 5 new tables, all using `t_ds_` prefix to fit the existing naming
convention, and place them in the new/upgrade SQL directory of
`dolphinscheduler-dao` (refer to the existing
`dolphinscheduler-dao/src/main/resources/sql/upgrade/<version>_schema/{mysql,postgresql}/dolphinscheduler_ddl.sql`
The upgrade script organization method).
```sql
-- Asset registry: describes a lake warehouse data asset that can be relied
upon
CREATE TABLE t_ds_asset (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
asset_key VARCHAR(512) NOT NULL, -- catalog.db.table[/partition_expr]
Normalized unique key
storage_format VARCHAR(32) NOT NULL, -- PAIMON / ICEBERG / HUDI
catalog_name VARCHAR(128) NOT NULL,
database_name VARCHAR(128) NOT NULL,
table_name VARCHAR(128) NOT NULL,
partition_expr VARCHAR(256),
description VARCHAR(512),
create_time DATETIME NOT NULL,
update_time DATETIME NOT NULL,
UNIQUE KEY uk_asset_key (asset_key)
);
--Original event table: only used as evidence to retain and troubleshoot,
not as a basis for scheduling decisions
CREATE TABLE t_ds_asset_event (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
event_id VARCHAR(128) NOT NULL,
source_type VARCHAR(16) NOT NULL, -- POLL / PUSH
asset_key VARCHAR(512) NOT NULL,
event_type VARCHAR(32) NOT NULL, -- SNAPSHOT_COMMITTED /
WATERMARK_ADVANCED / ...
snapshot_id BIGINT,
schema_id BIGINT,
watermark BIGINT,
commit_kind VARCHAR(32),
commit_time BIGINT,
delta_record_count BIGINT,
schema_changed TINYINT,
payload TEXT,
receive_time BIGINT NOT NULL,
create_time DATETIME NOT NULL,
UNIQUE KEY uk_dedup (source_type, asset_key, event_type, snapshot_id),
KEY idx_asset_snapshot (asset_key, snapshot_id),
KEY idx_event_type_time (event_type, receive_time)
);
-- Asset status table: the only source of truth for scheduling decisions
CREATE TABLE t_ds_asset_state (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
asset_key VARCHAR(512) NOT NULL,
latest_snapshot_id BIGINT,
latest_schema_id BIGINT,
latest_watermark BIGINT,
latest_tag VARCHAR(128),
quality_status VARCHAR(16) DEFAULT 'UNKNOWN',
schema_status VARCHAR(32) DEFAULT 'UNKNOWN',
backfill_status VARCHAR(16) DEFAULT 'NONE',
version BIGINT NOT NULL DEFAULT 0, -- optimistic locking
update_time DATETIME NOT NULL,
UNIQUE KEY uk_asset_key (asset_key)
);
-- Workflow/task dependency statement on assets
CREATE TABLE t_ds_asset_dependency (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
workflow_definition_code BIGINT NOT NULL,
task_definition_code BIGINT, -- empty means it will take effect on the
entire workflow (Strategy C)
asset_key VARCHAR(512) NOT NULL,
dependency_group VARCHAR(64) NOT NULL DEFAULT 'default', -- supports
multi-asset AND grouping
condition_json TEXT NOT NULL, -- see 3.6 Dependency Expressions
enabled TINYINT NOT NULL DEFAULT 1,
create_time DATETIME NOT NULL,
update_time DATETIME NOT NULL,
KEY idx_asset_key (asset_key),
KEY idx_workflow (workflow_definition_code)
);
-- Idempotent trigger history + auditing
CREATE TABLE t_ds_asset_trigger_history (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
trigger_key VARCHAR(256) NOT NULL, -- idempotent key, see 4.1
event_id VARCHAR(128),
asset_key VARCHAR(512) NOT NULL,
workflow_definition_code BIGINT NOT NULL,
workflow_instance_id BIGINT,
task_definition_code BIGINT,
task_instance_id BIGINT,
trigger_status VARCHAR(16) NOT NULL, -- TRIGGERING / SUCCESS / FAILED /
SKIPPED
reason VARCHAR(512),
create_time DATETIME NOT NULL,
update_time DATETIME NOT NULL,
UNIQUE KEY uk_trigger_key (trigger_key),
KEY idx_status_time (trigger_status, create_time)
);
--Event consumption site (used for Polling Scanner's breakpoint continued
scanning)
CREATE TABLE t_ds_asset_event_consumer_offset (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
source_type VARCHAR(16) NOT NULL,
asset_key VARCHAR(512) NOT NULL,
consumer_group VARCHAR(64) NOT NULL DEFAULT 'default',
last_snapshot_id BIGINT,
last_scan_time DATETIME,
UNIQUE KEY uk_offset (source_type, asset_key, consumer_group)
);
```
### 3.5 Scheduling trigger link (interconnected with the existing Command
mechanism)
```text
1. AssetEventScanner regularly scans Paimon $snapshots -> produces candidate
AssetEvents
2. Write t_ds_asset_event (use the unique constraint uk_dedup for insertion
and deduplication. If the conflict indicates that the event has been processed,
skip it directly)
3. After the event is successfully written, enter AssetState update:
- Read the current t_ds_asset_state (with version)
- If event.snapshotId <= state.latestSnapshotId, discard (out-of-order
protection, see 4.3)
- Otherwise, use optimistic locking UPDATE... WHERE asset_key=? AND
version=? Update status, version+1
4. After the status update is successful, trigger the DependencyResolver:
- Query the t_ds_asset_dependency associated with the asset_key
- For each dependency, evaluate according to condition_json (snapshot
exists/watermark>=X/quality=PASSED/schema=COMPATIBLE)
- AND aggregation of multiple dependencies under the same
dependency_group (the same workflow waits for multiple upstream assets to be
ready)
5. If judged as READY:
a. Calculate trigger_key (see 4.1) and make unique insertion into
t_ds_asset_trigger_history
- Insertion successful => Get the trigger right this time, continue to
step b
- Insertion failed (only conflict) => It means it has been triggered
by other concurrent Master/Scanner instances. Return directly without
triggering again.
b. Select the trigger method according to the dependency type:
- Strategy C (workflow level): Call existing
CommandService#createCommand,
Use the new CommandType (such as ASSET_EVENT_TRIGGER, or reuse
START_PROCESS and mark it in commandParam
triggerSource=ASSET_EVENT, assetKey, snapshotId and other audit
information),
The existing Command consumption link on the Master side completes
WorkflowInstance creation and DAG execution without changing the Master kernel.
- Strategy A (task level): If the workflow is already running and
contains the ASSET_SENSOR task,
Then update the dependency satisfaction mark of the corresponding
task instance, by the ASSET_SENSOR task (polling/subscribing to AssetState
changes at runtime)
Determine success by yourself and enter the existing task state
machine (reusing the same life cycle hook as DependentLogicTask).
c. Perform "native dependency access check" before actually triggering:
- If the user configures native dependencies that must be met first
(upstream task success, DEPENDENT condition, running window limit), the
verification will be passed before triggering;
- Only when the snapshot direct drive mode is explicitly configured,
asset version advancement can be directly used as the main trigger condition.
d. The trigger result (success/failure) is written back to
t_ds_asset_trigger_history.trigger_status,
Failed records are periodically scanned and retried by the
Compensation Scanner.
6. Master executes WorkflowInstance/TaskInstance according to the existing
normal link without knowing the source of the event.
```
### 3.6 Dependency expression (structured configuration, non-universal DSL)
```json
{
"assetKey": "paimon://catalog/db/ods_order",
"snapshotRequired": true,
"qualityStatus": "PASSED",
"schemaStatus": "COMPATIBLE"
}
```
`dependency_group` is used to express multi-asset AND semantics: only when
all dependencies under the same group are judged READY, the group is considered
ready; the workflow can be configured with multiple groups, and the groups are
OR (triggered when any group is ready). The semantics are consistent with
`DependentParameters.Dependence.relation` (AND/OR), making it easier for users
to understand migration costs. The first phase only supports AND-within-group /
OR-across-group two-level structures, and does not support general nested
Boolean expressions.
### 3.7 Idempotent design
Consists of `trigger_key` (see 4.1) + `t_ds_asset_trigger_history`. The only
constraint is the core guarantee of idempotence:
- The same workflowDefinitionCode + dependency_group + snapshot combination
of all related assets can only be successfully triggered once.
- When multiple Masters/Scanners scan the same batch of events concurrently,
"only one winner" is naturally realized through the unique constraints of the
database, without the need for distributed locks.
- If the call to `createCommand` fails (network/DB jitter), trigger_status
remains `TRIGGERING`, and the Compensation Scanner identifies the records that
have timed out but not finalized, and retry safely (before retrying, you need
to confirm whether the corresponding Command/WorkflowInstance has actually been
generated to avoid repeated creation. You can write back `trigger_key` in
`commandParam` and query whether there is an association before retrying.
Command/WorkflowInstance).
### 3.8 Multi-asset alignment (AND/OR)
- Dependencies are stored in `t_ds_asset_dependency` in `dependency_group`
groups. When triggering an evaluation, you need to read the entire set of
dependencies, evaluate `condition_json` one by one, and only enter the trigger
process when all are satisfied.
- To avoid the problem that asset B is not retriggered when it is ready
after "asset A is ready -> trigger evaluation -> asset B is not ready -> give
up": **When any asset event arrives, all dependency_groups associated with it
will be re-evaluated** (not just the asset itself), that is, the
DependencyResolver is driven by the `asset_key -> dependency_group` reverse
index to re-evaluate.
### 3.9 Out-of-order event processing
- Event level: `t_ds_asset_event` The only constraint is `(source_type,
asset_key, event_type, snapshot_id)`. Repeated reports are directly rejected by
the database and are regarded as successful deduplication.
- State level: When updating `AssetState`, strictly compare version numbers
(`snapshotId`/`watermark` monotonicity), and adopt the "only advance, no
rollback" strategy:
```sql
UPDATE t_ds_asset_state
SET latest_snapshot_id = ?, latest_watermark = GREATEST(latest_watermark,
?),
version = version + 1, update_time = ?
WHERE asset_key = ? AND version = ? AND ? > latest_snapshot_id;
```
If the number of rows affected by `UPDATE` is 0, it means that there is a
concurrency conflict or the event has expired (the old snapshot arrived late).
The event is marked `IGNORED` and the reason is recorded, which does not affect
the scheduling decision.
### 3.10 Failure retry and compensation
- **Compensation Scanner** (reuse the Polling Scanner infrastructure of 3.3,
independent scheduling cycle, for example, once every 5~15 minutes)
Responsibilities:
1. Compare Paimon's actual latest snapshot with
`t_ds_asset_state.latest_snapshot_id`. If it lags behind, it means there is a
missed collection event (Push failed or Scanner was temporarily down). Re-pull
the missing snapshot and fill in `t_ds_asset_event`.
2. Scan `t_ds_asset_trigger_history` for records that have been in the
`TRIGGERING`/`FAILED` state for a long time, and re-execute the trigger (first
check whether there is a corresponding Command/WorkflowInstance to avoid
repeated triggering).
3. For dependencies in the `BLOCKED_BY_*` state, a query interface can be
provided to explain "why it has not been triggered yet" (see Observability).
- The compensation link and the Push event share the same set of
deduplication and status update code paths to ensure that the two paths will
not be triggered twice.
## 4. Integration points with existing modules
### 4.1 How to declare asset dependencies in workflow/task definition
Two landing forms are recommended, and it is recommended to implement them
in stages:
**(1) Strategy A: New task type `ASSET_SENSOR` (recommended for MVP, minimum
risk)**
- Referring to the plug-in mechanism of the existing `DEPENDENT` task
(`DependentLogicTaskChannel` / `DependentLogicTaskChannelFactory` /
`DependentLogicTask`), add the `dolphinscheduler-task-asset-sensor` module
under `dolphinscheduler-task-plugin`:
- `AssetSensorParameters extends AbstractParameters`: Field structure
reference 3.6 Dependency expression JSON, plus `checkIntervalSeconds`,
`timeoutMinutes`.
- `AssetSensorLogicTask extends AbstractLogicTask<AssetSensorParameters>`:
Reuse the `onTaskRunning/onTaskPaused/onTaskKilled` life cycle of
`AbstractLogicTask`, and internally determine whether the conditions are met
through `AssetStateDao` polling (you can also register a callback, and the
DependencyResolver will actively push the READY status to reduce the polling
delay).
- The user adds an `ASSET_SENSOR` node in the workflow DAG as the "entry
sentinel task". The downstream tasks are connected in series according to the
ordinary DAG dependencies. After the task is successful, the DAG advances
backward normally.
- By default, it is recommended to place `ASSET_SENSOR` before the original
business task as a data readiness access control instead of replacing the
original DAG dependency; this can satisfy both "data version readiness" and
"user-configured native dependency".
- Advantages: Completely reuses the existing WorkflowInstance/TaskInstance
state machine, retry, timeout, alarm, UI display (task type registration
mechanism), with minimal changes, equivalent to "adding a new task plug-in to
DAG".
**(2) Strategy C: Workflow-level events directly create workflow instances
(for pure event-driven scenarios, second phase implementation)**
- Add `assetTriggerEnabled` and associated `dependency_group` in the
workflow definition extended attribute (you can reuse records with empty
`task_definition_code` in `t_ds_asset_dependency` to represent workflow-level
dependencies).
- After DependencyResolver determines READY, it calls
`CommandService#createCommand`, and `commandParam` carries trigger-related
audit fields such as `assetKey`, `snapshotId`, etc.
- Risk warning: It is necessary to clarify the corresponding relationship
between workflow instances and snapshot versions (see Section 11 Risks).
### 4.2 New interface in API layer
Add a new Controller in `dolphinscheduler-api` (refer to the existing
`ExecutorController` layering method: Controller -> Service -> DAO):
```text
POST /projects/{projectCode}/assets register/update assets
GET /projects/{projectCode}/assets Query the asset list
GET /projects/{projectCode}/assets/{assetKey}/state Query the current status
of assets
POST /projects/{projectCode}/asset-events External system proactively
reports events (Push mode)
GET /projects/{projectCode}/asset-events Query event history
(troubleshooting)
POST /projects/{projectCode}/asset-dependencies declare asset dependencies
for workflow/task
GET /projects/{projectCode}/asset-dependencies/{workflowDefinitionCode}
Query dependencies and their current determination status (why it is not
triggered)
GET /projects/{projectCode}/asset-trigger-history Query trigger history
(idempotent audit)
```
### 4.3 UI layer suggestions (not the focus of this article, brief
description)
- In the workflow definition canvas, the `ASSET_SENSOR` task node displays
the currently waiting assets, conditional expressions, and current AssetState
snapshot comparison (how many snapshots / how long is the watermark difference).
- Added "Asset Market" page: asset list, recent events, current status,
associated downstream workflows, trigger history timeline, similar to Airflow's
Dataset view.
## 4.4 Asset dependency model in DAG link
In the DAG of the new generation data warehouse, the triggering of all tasks
should follow the **pure snapshot model** without time expressions or native
dependency gates:
| DAG hierarchy | Dependency sources | Examples | Description |
|---|---|---|---|
| **ODS (data into the lake, first layer)** | Snapshot promotion of external
data sources | Flink CDC -> Paimon ods_order_latest snapshot | The original
data asset version source, usually snapshot events produced by external data
engines (Flink/Spark) |
| **DWD (dimension processing, middle layer)** | Depends on the snapshot of
the previous layer ODS | Waiting for ods_order_latest, ods_payment_latest
snapshots to be advanced | Multiple upstream assets are combined by AND
conditions |
| **DWS (topic aggregation, middle layer)** | Depend on multiple snapshots
of DWD | Waiting for dwd_trade_event, dwd_order_dim snapshot advancement | It
is also a multi-asset AND dependency, consistent with the DAG pre- and
post-dependencies in DAG |
| **ADS (External Service, Terminal Layer)** | Depends on DWS snapshot |
Waiting for dws_gmv_daily, dws_order_stat snapshot advancement | Final data
product layer |
**Key Principles**:
- The firing of any task in the entire DAG is driven entirely by the
snapshot version of its dependent assets, with no time expressions or cron
involved.
- The "pre- and post-dependencies" in the DAG are automatically converted
into "snapshot dependencies": that is, the latter task waits for the snapshot
produced by the previous task.
- Cross-stream dependencies (depending on tasks in other DAGs) are also
upgraded to the snapshot model: implemented by associating asset snapshots
produced by external DAGs.
- **No need to mix native dependencies and asset dependencies**: Under the
new generation data warehouse architecture, asset snapshot is the only trigger
basis.
## 4.5 Impact and transformation on the existing Master scheduling
architecture
### 4.5.1 Overview of the existing Master scheduling process
The core scheduling process of DolphinScheduler Master is as follows:
```text
1. CommandService regularly scans the t_ds_command table (incrementing by id)
↓
2. For each Command record:
a) Check commandType (START_PROCESS / SCHEDULER / COMPLEMENT_DATA / ...)
b) Read the associated workflow_definition (workflow definition)
c) Create WorkflowInstance (workflow instance, initial state = SUBMITTED)
d) Generate a DAG instance of the workflow (TaskInstance + TaskDependency)
↓
3. MasterScheduler regularly scans WorkflowInstance / TaskInstance (grouped
by status)
↓
4. Workflow instance status transfer:
SUBMITTED -> RUNNING (any non-skipped tasks have been submitted)
->SUCCESS (all tasks successful)
-> FAILURE (any critical path task failed)
↓
5. Task instance state transfer (core):
SUBMITTED -> READY/WAITING_DEPENDENCY (dependency is not satisfied)
-> RUNNING (all dependencies are met, sent to Worker)
-> SUCCESS/FAILED/...
↓
6. Dependence decision logic (current):
- DEPENDENT task: polling to query the status of upstream task instances
(whether SUCCESS)
- DAG front and rear dependencies: check whether the front task is SUCCESS
- Others: Judgment conditions are met immediately, READY directly
↓
7. Worker executes TaskInstance and reports the results
↓
8. Repeat steps 3-7 until the workflow is complete
```
### 4.5.2 Integration point of asset event scheduling
After the introduction of asset event-driven scheduling, the Master needs to
handle **two trigger sources** and **three-layer dependency determination**:
**Trigger source (Command creation)**:
```text
Original: START_PROCESS / SCHEDULER / COMPLEMENT_DATA / ...
↓
Enhanced: Add ASSET_EVENT_TRIGGER / or reuse START_PROCESS + special
commandParam
↓
The DependencyResolver module calls CommandService#createCommand
after determining that the asset is ready.
Write to t_ds_command (exactly the same as existing process)
```
**Three levels of dependency determination (from outside to inside)**:
```text
The first layer: workflow-level dependencies (Strategy C, second phase
implementation)
├─ Asset dependency determination: DependencyResolver evaluates
t_ds_asset_dependency (record with task_definition_code=null)
└─ Not satisfied => Do not create Command/workflow instance
Second level: task-level dependencies (Strategy A, MVP)
├─ Original DAG pre- and post-dependencies: Master’s existing logic, check
whether the pre-requisite tasks are SUCCESS
├─ ASSET_SENSOR task dependency: polling t_ds_asset_state, judgment
conditions (snapshot / watermark / quality)
└─ DEPENDENT task dependency: polling upstream task instance status
The third layer: workflow internal DAG dependencies
└─ Series/parallel connection between tasks, normal DAG topology execution
```
### 4.5.3 Specific modifications on the Master side
#### Transformation 1: DependencyResolver integrated into Master
**Conceptual Model**:
```java
public interface DependencyResolver {
/**
* Evaluate whether a dependency_group is READY
* @param workflowCode workflow definition encoding (used for
workflow-level dependency evaluation)
* @param taskCode task definition code (if null, indicates
workflow-level dependency)
* @param dependencyGroup dependency group name
* @return Whether the dependency is READY (true can be
triggered/released, false to continue waiting)
*/
DependencyStatus resolveDependency(Long workflowCode, Long taskCode,
String dependencyGroup);
/**
* Register a listener for asset status changes
* DependencyResolver can be internally based on event callbacks instead
of polling to reduce latency
*/
void registerAssetStateChangeListener(AssetStateChangeListener listener);
}
```
**Timing of workflow level trigger (Strategy C)**:
```text
Called immediately after the DependencyResolver determines that the
workflow-level dependency is READY:
commandService.createCommand(
Command.builder()
.commandType(CommandType.ASSET_EVENT_TRIGGER)
.workflowDefinitionCode(workflowCode)
.commandParam(JSON serialized {assetKey, snapshotId, triggerKey, ...})
.build()
);
然后在 t_ds_asset_trigger_history 中 INSERT 占位记录（trigger_key 唯一约束）。
Master 的现有 CommandService scanner 会照常消费这条 Command，创建 WorkflowInstance，执行 DAG。
无需在 Master 侧做特殊识别——完全复用现有流程。
```
#### 改造 2：ASSET_SENSOR 任务类型的执行生命周期
**在 Master 执行阶段的行为**：
```text
1. TaskInstance 状态 = SUBMITTED，判定是否可进入 READY
↓
2. MasterScheduler 调用 task plugin 的 tracker（AssetSensorTracker extends
TaskTracker）
↓
3. AssetSensorTracker.checkTaskStatus() 周期性调用
DependencyResolver#resolveDependency()
Return value:
- DependencyStatus.READY => 任务转为 READY，发送给 Worker 执行
- DependencyStatus.BLOCKED => 保持 WAITING_DEPENDENCY，继续轮询
- DependencyStatus.TIMEOUT => 任务转为 ERROR（超时失败）
↓
4. Worker 侧照常执行该任务（可能是 Shell / SQL，或空操作代表"依赖已满足"）
↓
5. 下游任务通过 DAG 前后置依赖关系正常推进
```
**代码集成点**（`dolphinscheduler-master` 模块）：
```text
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/
├── engine/
│ └── DAGExecutionEngine.java (现有，DAG 状态机)
│ ├── 改造点1：在 commitTask / dispatchTask 前检查 ASSET_SENSOR 任务
│ │ if (isAssetSensorTask) {
│ │ // 调用 DependencyResolver 检查依赖
│ │ if (!dependencyResolver.resolveDependency(...).isReady()) {
│ │ return; // 不提交给 Worker，保持 WAITING_DEPENDENCY
│ │ }
│ │ }
│ └── 改造点2：在工作流 SUBMITTED -> RUNNING 转移前
│ if (isAssetEventTrigger && hasWorkflowLevelAssetDependency) {
│ // 可选：在 Master 侧提前检查，尽早发现不满足的依赖
│ // （否则直接创建实例，在工作流执行时逐个任务检查）
│ }
├── processor/
│ ├── WorkflowProcessor.java (工作流实例处理)
│ │ └── 改造点3：处理 CommandType.ASSET_EVENT_TRIGGER
│ │ 解析 commandParam 提取 assetKey / snapshotId 等信息，
│ │ 存入 WorkflowInstance 的 commandParam 或扩展字段
│ │
│ └── TaskProcessor.java (任务实例处理)
│ └── 改造点4：任务状态转移时，对 ASSET_SENSOR 任务特殊处理
│ 调用 AssetSensorTracker.checkTaskStatus()
│
└── service/
├── DependencyResolverService.java (新增)
│ ├── 注入 AssetStateService (DAO 层)
│ ├── 注入 AssetDependencyService (DAO 层)
│ └── 实现 resolveDependency 逻辑
│ a) 查询 t_ds_asset_dependency(task_definition_code=?,
dependency_group=?)
│ b) 逐一查询 t_ds_asset_state 对应的 asset_key
│ c) 对每个资产条件求值（snapshot exists, watermark gte, quality status, ...）
│ d) 组合结果（同 group 所有条件 AND，不同 group 间 OR）
│
└── TrackerManager.java (现有，但需扩展)
└── 改造点5：注册 ASSET_SENSOR 对应的 tracker
trackerFactory.get(TaskType.ASSET_SENSOR)
=> new AssetSensorTracker(dependencyResolverService)
```
#### 改造 3：工作流实例启动与 SUBMITTED 状态的处理
**当前逻辑**（简化）：
```java
// MasterScheduler#submitWorkflow
Command command = commandService.findOne(commandId);
WorkflowInstance instance = new WorkflowInstance();
instance.setWorkflowDefinitionCode(command.getWorkflowDefinitionCode());
instance.setState(WorkflowExecutionStatus.SUBMITTED);
// ... 生成 TaskInstance，立即进入 DAG 执行
```
**新增逻辑（如果需要工作流级资产依赖检查）**：
```java
// 在 WorkflowInstance 创建后、生成 DAG 任务前
if (command.getCommandType() == CommandType.ASSET_EVENT_TRIGGER) {
// 可选：提前验证工作流级资产依赖是否已满足
// （这是一个检查点，如果仍未满足则可能是 Compensation 重试时的竞态）
boolean workflowLevelAssetReady = dependencyResolverService
.resolveDependency(workflowCode, null, "default")
.isReady();
if (!workflowLevelAssetReady) {
// 记录告警，可选地延迟创建 TaskInstance
log.warn("Workflow {} triggered by asset event but asset dependency
no longer ready. " +
"This may indicate compensation retry or state backtrack.",
workflowCode);
}
}
// 正常流程继续，生成 DAG 任务
for (TaskDefinition taskDef : dag.getTasks()) {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setTaskDefinitionCode(taskDef.getCode());
taskInstance.setState(TaskExecutionStatus.SUBMITTED);
if (isAssetSensorTask(taskDef)) {
// ASSET_SENSOR 任务状态初始化为 WAITING_DEPENDENCY
taskInstance.setState(TaskExecutionStatus.WAITING_DEPENDENCY);
}
// ... 保存 taskInstance
}
```
#### 改造 4：Master 的循环扫描逻辑
**原有扫描周期**：
```text
CommandService.scanner() // 扫描 t_ds_command，周期=配置（默认 5s）
↓
WorkflowProcessor.processWorkflow() // 处理工作流状态转移
↓
TaskProcessor.processTask() // 处理任务状态转移，检查依赖
↓
AbstractTaskTracker.checkTaskStatus() // 调用具体 task plugin 的 tracker（如
DependentTaskTracker）
```
**新增扫描逻辑**（与现有不冲突）：
```text
新增 AssetEventCompensationScanner（可选，作为 Master 内部的后台线程）
周期：5~15 分钟（独立于 CommandService 扫描）
↓
扫描 t_ds_asset_event 和 t_ds_asset_state，检查：
1. 是否有漏采集的事件（e.g. Paimon 实际最新 snapshot_id >
t_ds_asset_state.latest_snapshot_id）
2. 是否有卡顿的 trigger_history（TRIGGERING 或 FAILED 状态超过 N 分钟）
↓
对于漏采集：补写 t_ds_asset_event（去重自动处理）-> 触发 DependencyResolver 重新评估
对于卡顿：可选地重试 createCommand（检查唯一约束冲突，确保不重复）
↓
该扫描是"兜底补偿"，不影响主调度流程
```
### 4.5.4 对现有架构的最小化影响
**不需要改动**：
- ✅ WorkflowInstance 和 TaskInstance 的核心状态机（SUBMITTED / RUNNING / SUCCESS /
FAILED / ...）
- ✅ DAG 前后置依赖的表达和执行（保持原样）
- ✅ DEPENDENT 任务的现有逻辑（完全独立，不干扰）
- ✅ Worker 任务执行的生命周期（无感知）
- ✅ Alert / Monitor / Backfill 等周边功能（兼容）
**需要改动**（但都是插件化、非侵入式）：
- 新增 `ASSET_SENSOR` 任务类型（如 `DependentLogicTask` 一样作为 task plugin）
- 新增 `AssetSensorTracker` 作为该任务的 tracker（实现 TaskTracker 接口）
- 新增 `DependencyResolverService` 负责依赖评估逻辑（Service 层，不改 Master 内核）
- 新增 `CommandType.ASSET_EVENT_TRIGGER` 类型（可选，或复用 START_PROCESS）
- 在 `TaskProcessor#submitTask()` 前增加一行检查：
```java
if (task is ASSET_SENSOR) {
if (!dependencyResolver.resolveDependency(...).isReady()) {
return; // 不提交
}
}
```
**改动的代码行数**：
预估 **< 500 行**（包括新增的 DependencyResolverService、AssetSensorTracker
和必要的集成点检查），相对于 DolphinScheduler 整体代码量（数万行）可以说是非常小的改动。
### 4.5.5 并发与一致性保证
**多 Master 并发场景**：
```text
场景：5 个 Master 实例同时运行，都在扫描 t_ds_command 和执行工作流
冲突点1：CommandService.createCommand() 写入同一条 Command
=> 依赖 t_ds_command 的 PK 或 sequence，一般由应用侧写入，不会重复
冲突点2：多个 Master 同时评估 ASSET_SENSOR 任务的依赖，决定是否提交给 Worker
=> 每个 Master 各自维护 TaskTracker，评估结果本地缓存
=> 当依赖首次从 NOT_READY 转为 READY 时，最快的 Master 抢到 Worker 资源
=> TaskProcessor#submitTask() 会检查 TaskInstance 是否已有 Worker 分配，重复提交会被幂等处理
=> 无需分布式锁，因为 TaskInstance 表会做状态检查
冲突点3：多个 Master 同时插入 t_ds_asset_trigger_history
=> 唯一约束 uk_trigger_key 保证只有一个成功
=> 失败的 Master 捕获唯一冲突异常，记录日志后继续（不影响调度）
```
**幂等性保证**：
```text
trigger_key 形如：
workflow_12345 + dependency_group_default + snapshot_1001 + snapshot_1002
+ ...
= 完整标识"该工作流依赖这一组快照组合"的唯一键
同一 trigger_key 最多被成功创建一次 t_ds_asset_trigger_history 记录。
对应的 Command 也最多被创建一次（或可在 commandParam 中埋入 trigger_key 做检查）。
因此，无论有多少个 Master 并发评估，最终只会创建一个 WorkflowInstance。
```
### 4.5.6 关键配置项
Master 侧需要的新配置（写入 Master 的 application.yaml）：
```yaml
# 资产事件驱动调度开关
dolphinscheduler:
asset-event-scheduling:
enabled: true # 全局开关
# DependencyResolver 配置
dependency-resolver:
cache-expire-seconds: 60 # 资产状态本地缓存过期时间
resolve-timeout-seconds: 5 # 单次依赖评估的超时时间
# AssetEventCompensationScanner 配置
compensation-scanner:
enabled: true
interval-seconds: 300 # 补偿扫描周期（5分钟）
batch-size: 100 # 每次扫描的批量大小
# 事件源配置
event-source:
paimon-polling:
enabled: true
interval-seconds: 30 # Paimon $snapshots 轮询周期
batch-size: 500
push-receiver:
enabled: true # 是否启用 HTTP/Webhook 接收事件
```
这些配置都是**可选的、渐进式的**，MVP 时可以只启用 `enabled: true`，使用默认值。
---
## 5. 端到端示例：基于 Paimon 订单表的 GMV 场景
### 5.1 业务流程与数据资产
```text
MySQL 订单库（order / payment / refund 表）
-> Flink CDC 持续消费 binlog，生产 Paimon snapshot
-> Paimon 湖仓表：
ods_order_latest（主键表，最新订单状态）→ snapshot推进
ods_payment_latest（主键表，最新支付状态）→ snapshot推进
-> DWD 层任务：dwd_trade_event_fact（识别支付/退款事件）
依赖 ods_order_latest + ods_payment_latest 快照都推进
-> DWS 层任务：dws_gmv_daily（每日GMV聚合）
依赖 dwd_trade_event_fact 快照推进
-> ADS 层任务：ads_gmv_dashboard（对外报表/API）
依赖 dws_gmv_daily 快照推进
```
### 5.2 工作流定义（DAG 结构）——纯snapshot模型
**新一代数仓DAG结构**（无cron、无时间表达式）：
```text
Flink CDC 持续运行，生产 ods_order_latest、ods_payment_latest snapshot
↓
DAG工作流启动（由外部系统触发，例如Airflow或DolphinScheduler的API）
↓
Task 1: DWD_trade_event_fact
依赖：ods_order_latest snapshot_id >= N1 AND ods_payment_latest snapshot_id
>= N2
执行：读两张表的最新快照，识别交易事件
产出：dwd_trade_event_fact snapshot
↓
Task 2: DWS_gmv_daily
依赖：dwd_trade_event_fact snapshot_id >= M1
执行：聚合每日GMV
产出：dws_gmv_daily snapshot
↓
Task 3: ADS_gmv_dashboard
依赖：dws_gmv_daily snapshot_id >= P1
执行：同步到报表库
产出：ads_gmv_dashboard 就绪
```
**核心改进**：
- 完全无cron，无固定执行时刻。 DAG启动后，每个任务的执行取决于其依赖快照的推进。
- 如果Flink CDC延迟，ODS快照推进慢，则DWD会等待；一旦快照到达，DWD立即执行。
- 多个上游资产（ods_order_latest + ods_payment_latest）的快照通过AND条件组合。
### 5.3 资产依赖声明示例
**DWD 层对两个 ODS 表的依赖**：
```json
[
{
"workflowDefinitionCode": 12345,
"taskDefinitionCode": 100, // DWD_trade_event_fact 任务
"assetKey": "paimon://ods_db/ods_order_latest",
"dependencyGroup": "trade_event_group",
"conditionJson": {
"assetKey": "paimon://ods_db/ods_order_latest",
"snapshotRequired": true,
"qualityStatus": "PASSED",
"schemaStatus": "COMPATIBLE"
},
"enabledFlag": 1
},
{
"workflowDefinitionCode": 12345,
"taskDefinitionCode": 100, // 同一任务的第二个依赖
"assetKey": "paimon://ods_db/ods_payment_latest",
"dependencyGroup": "trade_event_group",
"conditionJson": {
"assetKey": "paimon://ods_db/ods_payment_latest",
"snapshotRequired": true,
"qualityStatus": "PASSED",
"schemaStatus": "COMPATIBLE"
},
"enabledFlag": 1
}
]
```
**DWS 层对 DWD 层的依赖**：
```json
[
{
"workflowDefinitionCode": 12345,
"taskDefinitionCode": 200, // DWS_gmv_daily 任务
"assetKey": "paimon://dw_db/dwd_trade_event_fact",
"dependencyGroup": "gmv_group",
"conditionJson": {
"assetKey": "paimon://dw_db/dwd_trade_event_fact",
"snapshotRequired": true,
"qualityStatus": "PASSED"
},
"enabledFlag": 1
}
]
```
**触发行为**：
1. Flink CDC 持续消费 MySQL binlog，每次产生新 snapshot 时会生产事件（例如 ods_order_latest
snapshot_id=1001）。
2. AssetEventScanner 或 Push 接口捕获事件，写入 `t_ds_asset_event`。
3. AssetStateService 以乐观锁更新
`t_ds_asset_state`：`asset_key=paimon://ods_db/ods_order_latest,
latest_snapshot_id=1001`。
4. DependencyResolver 反向索引扫描：该 asset_key 关联的所有 dependencyGroup 有哪些？ （例如
trade_event_group）。
5. 对 trade_event_group 中的所有依赖进行评估：
- ods_order_latest 快照存在? ✓
- ods_payment_latest 快照存在? ✓ (假设也已推进)
- 两者 qualityStatus 都=PASSED? ✓
→ 依赖组 READY
6. 计算
`trigger_key=workflow_12345_taskdef_100_group_trade_event_group_snapshot_1001_1002`（包含所有关键快照ID），向
`t_ds_asset_trigger_history` 唯一插入。
7. 插入成功 → DWD_trade_event_fact 任务实例被通知依赖满足，立即执行。
8. DWD 产出新快照（例如 dwd_trade_event_fact snapshot_id=2001），重复步骤2-7，触发 DWS。
9. ADS 类似地等待 DWS 快照推进。
## 6. 观测性设计
### 6.1 关键指标（Metrics）
系统应暴露以下 Prometheus 指标（Micrometer），方便 Grafana 大盘接入：
```text
# 事件接入层
lakehouse_asset_event_received_total{source_type="paimon", asset_key="..."}
- 意义：从指定资产收到的事件总数
lakehouse_asset_event_deduplicated_total{source_type="paimon",
asset_key="..."}
- 意义：被去重丢弃的重复事件数
lakehouse_asset_event_invalid_total{source_type="paimon", asset_key="...",
reason="xxx"}
- 意义：被拒绝的无效事件（格式错误、资产不存在等），按原因分类
# 资产状态层
lakehouse_asset_state_updated_total{asset_key="..."}
- 意义：资产状态被推进的次数
lakehouse_asset_state_lag_seconds{asset_key="..."}
- 意义：资产最新快照的年龄（当前时间 - snapshot commit_time）
lakehouse_asset_watermark_lag_seconds{asset_key="..."}
- 意义：资产 watermark 相对于业务日期的延迟（例如 watermark=10-10 23:59:59，当前日期 10-11，则
lag=-10秒表示超前）
# 依赖与触发层
lakehouse_asset_dependency_ready_total{asset_key="...", workflow_code="...",
dependency_group="..."}
- 意义：依赖就绪的次数
lakehouse_asset_dependency_blocked_by{asset_key="...", workflow_code="...",
blocked_by="snapshot|watermark|quality|schema"}
- 意义：依赖被卡住的原因分布
lakehouse_asset_trigger_success_total{workflow_code="...",
trigger_source="asset_event"}
- 意义：由资产事件成功触发的工作流实例数
lakehouse_asset_trigger_failed_total{workflow_code="...", reason="xxx"}
- 意义：触发失败的工作流数，按原因分类（command_create_failed、duplicate_trigger 等）
# 补偿层
lakehouse_asset_compensation_scan_total{asset_key="..."}
- 意义：补偿扫描器运行次数
lakehouse_asset_compensation_missed_events{asset_key="..."}
- 意义：补偿发现的漏采集事件数
lakehouse_asset_trigger_history_retry_total{trigger_status="..."}
- 意义：触发历史被重试的次数
```
### 6.2 关键日志字段
所有日志输出应包含以下上下文字段，便于链路追踪和问题排查：
```json
{
"timestamp": "2024-09-10T10:15:30.123Z",
"level": "INFO",
"logger": "org.apache.dolphinscheduler.service.asset.AssetEventScanner",
"traceId": "abc123xyz",
"assetKey": "paimon://ods_db/ods_order_latest",
"partitionKey": "dt=20240910",
"snapshotId": 1001,
"schemaId": 5,
"watermark": "2024-09-10T23:59:59Z",
"eventType": "SNAPSHOT_COMMITTED",
"eventId": "paimon_evt_20240910_1001_xxx",
"workflowDefinitionCode": 12345,
"taskDefinitionCode": 54321,
"workflowInstanceId": 67890,
"taskInstanceId": 11111,
"triggerKey": "workflow_12345_group_gmv_default_snapshot_1001",
"status": "READY | BLOCKED_BY_QUALITY | TRIGGERED | FAILED",
"reason": "Asset quality check passed, dependency READY",
"message": "Asset snapshot processed, dependency READY, triggering
workflow instance..."
}
```
### 6.3 告警规则（Alerting Rules）
| 告警条件 | 阈值 | 优先级 | 处理建议 |
|---|---|---|---|
| `lakehouse_asset_state_lag_seconds > 3600` | 1 小时 | P2 | 检查 Paimon
表是否有新快照产生；检查 Scanner 健康状态 |
| `lakehouse_asset_watermark_lag_seconds > 600` | 10 分钟 | P2 | 检查数据源
CDC/Flink 是否延迟；检查 Paimon 表的 `_watermark` 字段 |
| `lakehouse_asset_trigger_failed_total increase > 3/5min` | 5 分钟内失败 3 次 |
P1 | 检查 Command 创建是否异常；检查 Master 可用性 |
| `lakehouse_asset_dependency_blocked_by{blocked_by="quality"} > 0` 持续 >
30min | 30 分钟 | P2 | 质量检查失败，需要数据生产方或质量团队介入 |
| `lakehouse_asset_compensation_missed_events > 10/hour` | 1 小时 > 10 个 | P1
| 说明 Push 或 Poll 路径有系统问题，补偿机制在弥补但需要根本修复 |
## 7. 资产标识规范
### 7.1 AssetKey 命名规范
新一代数仓中，assetKey 标识一个湖仓表的全局唯一位置，无需包含分区变量：
```text
paimon://<catalog_name>/<database_name>/<table_name>
```
Example:
- `paimon://ods_db/ods_order_latest` — ODS层订单表
- `paimon://dw_db/dwd_trade_event_fact` — DWD层交易事件表
- `paimon://dws_db/dws_gmv_daily` — DWS层每日GMV表
**注意**：
- assetKey
指向表级别，不包含分区（partition_spec），因为新一代数仓中每个表的所有分区都在同一个Paimon/Iceberg/Hudi主键表中，版本推进是表级的。
- 若需要追踪不同分区的快照独立性，应在 `t_ds_asset_event` 中的 `partitionKey` 字段记录（例如
`dt=20240910`）。
### 7.2 AssetKey 注册与发现
- 资产需在 `t_ds_asset` 表中注册（API或批量脚本），包含
sourceType（paimon/iceberg/hudi）、catalog、database、table 等元信息。
- 工作流定义依赖时，通过 UI"资产大盘"或 API 查询并选择已有资产，类似 Airflow Dataset 选择器。
- 若依赖的 assetKey 不存在，DependencyResolver 会记录警告并返回
BLOCKED_BY_UNKNOWN_ASSET，防止工作流被永久卡住。
## 8. 分阶段实施路线图
### 阶段一（MVP，2~3 周量级）：Paimon snapshot -> ASSET_SENSOR 单资产触发闭环
-
新增表：`t_ds_asset`、`t_ds_asset_event`、`t_ds_asset_state`、`t_ds_asset_dependency`（仅
`task_definition_code`
非空场景）、`t_ds_asset_trigger_history`、`t_ds_asset_event_consumer_offset`。
- 实现 `AssetEventScanner`（轮询 Paimon `$snapshots`）。
- 实现 `AssetStateService`（乐观锁更新、乱序丢弃）。
- 实现 `DependencyResolver`（仅支持单资产、snapshot 存在性判定）。
- 实现 `ASSET_SENSOR` 任务插件（对接 Strategy A）。
- 明确 MVP 的问题边界：优先解决"任务需要按 snapshot 就绪触发而不是时间触发"的场景，不改变既有原生 DAG 依赖语义。
- 打通端到端：手工制造 Paimon 表 snapshot 推进 -> 观察 ASSET_SENSOR 任务在几十秒内由等待转为成功 ->
下游任务执行。
- 补齐幂等/乱序/基础可观测（日志字段，见 5.3）单元测试。
关键接口草图：
```java
public interface AssetEventSource {
List<AssetEvent> pollNewEvents(Asset asset, Long sinceSnapshotId);
}
public interface AssetStateService {
/** 乱序保护 + 乐观锁更新，返回 true 表示状态被推进 */
boolean applyEvent(AssetEvent event);
AssetState getState(String assetKey);
}
public interface AssetDependencyResolver {
/** 返回 READY / BLOCKED_BY_WATERMARK / BLOCKED_BY_QUALITY /
BLOCKED_BY_SCHEMA / WAITING */
DependencyEvalResult evaluate(AssetDependency dependency);
}
public interface AssetTriggerService {
/** 幂等触发，内部完成 trigger_key 唯一插入 + Command 创建或 ASSET_SENSOR 状态回写 */
TriggerResult triggerIfReady(AssetDependencyGroup group);
}
```
### 阶段二：多资产 AND/OR、watermark/quality/schema 条件、Compensation Scanner
- 扩展 `condition_json` 支持 watermark/qualityStatus/schemaStatus 判定。
- 支持 `dependency_group` 多资产 AND 聚合、跨 group OR。
- 上线 Compensation Scanner，定期核对 Paimon 实际快照与 AssetState 差异、重试失败触发。
- 补充观测指标与 Grafana 面板。
### 阶段三：Push 上报接口、工作流级 Strategy C、Iceberg/Hudi 事件源接入
- 新增 `POST /asset-events` API，Flink/Spark 作业可主动上报，与 Polling 共存去重。
- 支持工作流级直接创建 WorkflowInstance（Strategy C），并补充 bizDate 推导与补数场景边界的规则文档。
- 扩展 `AssetEventSource` 接口的 Iceberg/Hudi 实现。
### 阶段四：治理与观测增强（非本次核心，预留）
- 资产血缘、资产全生命周期看板、跨项目资产依赖等。
## 9. 新一代数仓DAG的改造与兼容性
### 9.1 与现有 cron 调度的关系
新一代数仓DAG**完全基于snapshot版本推进**，而不是时间触发。 This means:
- **不保留 cron**：在新架构下，DAG 不再有"定时执行时间"的概念。
- **不保留时间依赖**：DAG 中的任务也不再依赖原始的 DEPENDENT（它面向历史实例结果）或"业务日期"这类时间维度。
- **完全由数据资产版本驱动**：DAG 启动后，每个任务的执行 100% 取决于其上游资产快照的推进。
对于现有的使用者：
- 若要升级到新一代数仓架构，需要**全量重新定义 DAG**，将所有"cron + DEPENDENT"的组织方式改为"snapshot +
assetKey"。
- 这是一次架构升级，不是渐进式改造，因此需要统一规划和实施。
### 9.2 ASSET_SENSOR 任务类型的注册
- ASSET_SENSOR 是新增的任务类型，需要在 `dolphinscheduler-task-plugin` 注册并打包发布。
- 现有集群升级到新版本后自动支持。
- 若工作流定义中引用了 ASSET_SENSOR 而集群未安装，Master 会记录错误并跳过该任务。
### 9.3 数据库升级路径
- `t_ds_asset` 等新表通过 `dolphinscheduler-dao` 的 SQL migration 脚本自动创建。
- 升级无需停机，migration 脚本幂等且对现有数据无侵入。
- 兼容 MySQL 5.7+、PostgreSQL 10+ 等常见数据库。
### 9.4 与既有 DependentTask 的区别
| 特性 | DEPENDENT 任务（旧） | ASSET_SENSOR 任务（新） |
|---|---|---|
| 依赖对象 | 上游任务/工作流实例执行结果 | 外部数据资产快照版本 |
| 适用 | 传统DAG内的任务编排 | 新一代数仓的数据版本驱动 |
| 触发条件 | 上游任务成功/失败 | 资产快照推进/质量通过 |
## 10. 常见问题解答（FAQ）
**Q1：DAG中如何处理多个上游资产（例如DWS依赖多个DWD表）的依赖？ **
A1：通过 `dependencyGroup` 和AND条件组合。同一 dependencyGroup
下的所有资产依赖都必须READY，该组才READY。 For example:
```json
[
{ "taskDefinitionCode": 200, "assetKey": "paimon://dw_db/dwd_trade_event",
"dependencyGroup": "dws_gmv" },
{ "taskDefinitionCode": 200, "assetKey": "paimon://dw_db/dwd_order_dim",
"dependencyGroup": "dws_gmv" }
]
```
两张DWD表的快照都推进时，DWS才会触发。
---
**Q2：如果资产事件丢失了（例如 Paimon snapshot 没有被 Scanner 捕获），下游会永久卡住吗？ **
A2：不会。有两层防护：
1. 每个任务都设置了 `timeoutMinutes`（例如 60 分钟），超时后任务进入 ERROR 状态，触发告警。
2. Compensation Scanner 每 5~15 分钟运行一次，对比 Paimon 实际快照与
AssetState，若发现漏采集会补写事件并重新触发。
---
**Q3：多个工作流依赖同一个资产，它们会被重复触发吗？ **
A3：不会。每个工作流对该资产的依赖是独立的 trigger_key，例如：
```text
workflow_A_group_default_snapshot_1001
workflow_B_group_default_snapshot_1001
```
虽然都依赖同一个 snapshot，但触发记录是分开的，各自触发一次。
---
**Q4：如果任务因网络原因多次重试，会重复触发下游吗？ **
A4：不会。任务实例的重试机制在 Worker 侧处理，只要任务最终成功，就会释放下游任务。下游任务会通过 DAG
依赖被触发，同样遵循"任务实例唯一性"原则，不会重复。
---
**Q5：是否支持多个数据湖表格式（Paimon、Iceberg、Hudi）的混合依赖？ **
A5：MVP（阶段一）只支持 Paimon。阶段二扩展 watermark/quality/schema 条件后可统一处理。阶段三会增加
Iceberg/Hudi 事件源，理论上支持混合依赖，但需要确保每种格式的 EventSource 实现质量和测试覆盖。
---
**Q6：如何在新架构下支持跨流依赖（DAG之间的任务依赖）？ **
A6：跨流依赖升级为"依赖外部DAG产出的资产快照"。例如，DAG-B 的某个任务依赖 DAG-A 产出的某个表快照：
```json
{
"assetKey": "paimon://dw_db/external_dag_a_output_table",
"dependencyGroup": "cross_flow",
"conditionJson": {
"snapshotRequired": true,
"qualityStatus": "PASSED"
}
}
```
这样 DAG-B 会等待外部 DAG-A 的表快照推进，而不是等待任务实例。
---
**Q7：数据库如果 INSERT 唯一约束冲突时的异常处理，会影响性能吗？ **
A7：数据库唯一约束冲突通常很快（< 1ms），是预期内的正常路径，被捕获后记录日志继续执行，不会有明显性能影响。但如果并发度极高（例如 10+
Master 实例同时竞争触发），建议监控数据库连接池和索引性能。
## 11. 风险与开放问题
1.
**补数（backfill）与事件驱动的语义定义**：手动补数会重新产出历史快照，需要明确这些历史快照是否应触发下游任务。建议在设计中定义"backfill_mode"标记，告诉系统是否对历史快照触发。
2. **多 Master 并发触发的竞态窗口**：虽然唯一约束保证最终只有一次成功插入 `trigger_history`，但在
`createCommand` 调用与 `trigger_history` 落库之间仍存在短暂不一致窗口。需要明确"先插入 trigger_history
占位再调用 createCommand"的顺序约束，并对 `createCommand` 失败做补偿。
3. **任务轮询频率与数据库压力**：如果同时存在大量任务实例轮询同一批资产状态，需要考虑批量拉取/缓存策略，避免对
`t_ds_asset_state` 造成过大查询压力。建议引入 DependencyResolver 主动回调机制替代逐任务轮询。
4. **CommandType 语义扩展的兼容性**：新增 `ASSET_EVENT_TRIGGER` 类型需要评估对现有依赖
`CommandType` 做 switch-case 穷举的代码路径（如告警、日志、UI 展示）的影响。需全仓库搜索所有 `CommandType`
的使用点做兼容适配。
5. **跨 catalog/跨项目资产依赖的权限模型**：资产可能归属不同 DolphinScheduler
项目甚至不同租户，依赖声明与触发是否需要跨项目鉴权尚未设计。需要后续单独讨论。
6. **事件从 Push 与 Poll 双路径去重的时钟/顺序依赖**：两条路径若在极短时间内先后到达同一 snapshot
事件，唯一约束能防止重复落库，但需要验证在高并发下数据库唯一索引冲突处理的实现正确性与性能。
7. **exactly-once 声明范围**：本设计只能保证"同一 trigger_key 的触发记录只成功创建一次"，不能保证"下游
Command 消费与 WorkflowInstance 创建"在极端故障场景下的严格 exactly-once。需要在文档中明确边界。
8. **分区与无分区表的混合依赖**：当DAG中某个任务同时依赖有分区表和无分区表的快照时，如何统一表达和评估。需要在依赖表达式中定义clear的规则。
## 12. 总结
本设计将 DolphinScheduler 从"时间DAG调度系统"升级为"数据资产版本驱动的调度系统"。核心原理是：
- **整个工作流（DAG）代表ODS→DWD→DWS→ADS的完整数据加工链路**。
- **DAG中每一层任务的触发完全由其依赖资产的快照版本推进决定**，而非时间或手工触发。
- **通过新增 ASSET_SENSOR 任务类型**，复用现有任务实例状态机和DAG执行机制。
- **通过新增资产表和事件去重机制**，保证幂等触发和多Master并发安全。
- **不破坏现有 DolphinScheduler 内核**，以插件化任务和现有 Command 入口实现对接。
这是面向新一代数仓（Paimon/Iceberg/Hudi）的原生调度方案，解决了传统时间触发与数据可用性错配的根本问题。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
