[tiankongdeguiji]

Summary

Supports an online-learning setup where a daily incremental job is the base and a streaming (Kafka) job continues from each daily checkpoint. Each day the streaming job built on day T's checkpoint starts alongside the one built on day T-1. Once the new one has caught up, the old one should retire and EAS should switch over.

• New env ONLINE_LEARNING_SUCCESSOR_DIR (the next lineage's model_dir). Every log_step_count_steps, all ranks gather their consumed event-times and reconcile them with the same quorum as checkpoint meta data_ts. Rank 0 writes the result to <model_dir>/progress.json, reads the successor's, and broadcasts the decision. When the successor trails by no more than ONLINE_LEARNING_HANDOVER_TOLERANCE_SECS (default 300 s), all ranks leave both training loops at the same step.
• On hand-over the epoch-end and final checkpoints are skipped, because a retired lineage is never resumed. The trailing delta dump, upload drain and online dense finalize_publish still run. Other termination paths are unchanged.
• At startup it fails fast if the successor model_dir equals this run's own, or if the tolerance is not a finite non-negative number. Both are checked on every rank, so a bad value can't strand the other ranks.
• tzrec.export gains --edit_config_json (same semantics as train_eval / predict), so the full export can set the per-lineage FeatureDB version without a config placeholder.
• Version bumped to 1.5.6.
• New best-practice doc docs/source/best_practices/online_learning.md. It covers the DataWorks DAG (train_daily -> full_export -> train_stream -> deploy) with dlc submit node commands, a timeline diagram, per-lineage isolation (Kafka group.id, FeatureDB version, model and serving dirs), and the PyODPS 3 deploy node (eascmd modify).

Design choices

• Progress signal: a small progress.json in each lineage's model_dir, independent of online dense export and of the dataset type (any source that sets the data timestamp). An earlier revision compared the published dense_hot_export/current.json. That tied hand-over to ONLINE_DENSE_EXPORT=1, and its value only moved once per publish interval.
• Where it lives: OnlineLearningHandover and create_online_learning_handover in tzrec/utils/online_learning_util.py (progress file, quorum, tolerance, env parsing). main.py only creates it, calls update_and_check() at the log cadence, and acts on the result by exiting both loops and running the existing shutdown. This matches how DeltaEmbeddingDumper / OnlineDenseExportManager are wired. Neither the checkpoint nor the export manager is involved. A source without timestamps never hands over, and rank 0 warns once.
• Catch-up is approximate on purpose: quorum consumed freshness, with no per-partition offset comparison or final-watermark handshake. With dense export on, the successor's served version can trail its consumption by about one publish interval at the switch, which is accepted.
• Tolerance: once both lineages are at the Kafka head they consume at the same rate. The successor's value can then trail by one check interval plus partition skew, so a strict >= could fail indefinitely. That would keep the old job running until its max runtime and block every later deploy. A fixed 300 s default covers this, because progress now updates every check rather than every publish. Rank 0 logs the observed lag at each check.
• Switch trigger: the predecessor retires itself, and the DataWorks deploy node depends on the previous cycle's stream node, so no extra polling job is needed. Deploy does not repeat the freshness comparison. Stream jobs therefore use num_epochs: 1 and no num_steps, since any normal exit triggers deploy. A self-dependency on the deploy node keeps switches in date order.

The deploy template's Processor keys were verified against the TorchRecProcessor source. The service mounts the daily full export as the model dir and the lineage serving root separately, and sets dense_hot_load.current_json explicitly, because the default looks under the model dir. It sets feature_store_embedding_polling.{feature_view_name, version: ol_<bizdate>}, which defaults to apply mode, and distributed_embedding: true.

Test Plan

• python -m unittest tzrec.utils.online_learning_util_test tzrec.main_test: 15 + 38 tests pass. New cases cover:
  • the decision: ahead, equal, behind, within/beyond tolerance; successor progress missing, corrupt or non-object; timestamps that are -1, NaN, inf, non-numeric or bool; own progress content
  • no event-time: no progress written, no hand-over, a single warning
  • setup: no successor -> disabled; no online export needed; default 300 s; env override and its validation; identical model_dir rejected
  • a mocked train loop: caught_up is called at the log cadence with the step and batch timestamp, and hand-over skips the epoch-end and final saves while dump, drain and finalize still run
  • tzrec.export forwarding --edit_config_json
• Mutation check: removing the outer-loop break makes the retirement test fail on the epoch-end save.
• 2-rank torchrun on 2x A10 (NCCL), ranks with different timestamps:
  • the quorum value (900 with quorum 1.0) is written by rank 0 only, with no temp file left behind;
  • no hand-over while the successor's progress is absent;
  • both ranks hand over once it trails by 50 s with a 60 s tolerance.
• pre-commit run, pyrefly check: clean.
• Not run: an end-to-end streaming run against a live Kafka.

🤖 Generated with Claude Code

https://claude.ai/code/session_01FouSwvnwPn2Z61mLnv4vsH

[github-actions]

These object collectives run on the default process group, which is NCCL on GPU training. all_gather_object/broadcast_object_list then pickle to CPU tensors, copy them H2D/D2H through torch.cuda.current_device(), and force a device sync in the train loop — on the same communicator the forward/backward collectives use. The sibling collectives in this same loop deliberately avoid that: CheckpointManager gathers event-times on a dedicated gloo group (checkpoint_util.py:463) and OnlineDenseExportManager does the same (online_dense_export_util.py:359), and dist_util.get_dist_object_pg() exists for exactly this. Worth reusing one of those patterns here (e.g. create a gloo side group in __init__ and pass group= to both calls). Amortized cost is small at the default log_step_count_steps, but it scales linearly if the log cadence is lowered.

[github-actions]

Minor: a JSON integer too large for a float (e.g. a 400-digit data_ts written by hand or by a foreign writer on the shared mount) makes math.isfinite(ts) raise OverflowError (it converts the int to float), which is not caught by except (OSError, ValueError) in _read_successor_ts — so instead of degrading to "no hand-over" like the other corrupt-file cases, rank 0 dies between the gather and the broadcast and the job goes down with a confusing traceback. Converting inside a try: float(ts) except (TypeError, ValueError, OverflowError): return None (or adding OverflowError to the read's except tuple) closes it in one line.

[github-actions]

Nice coverage of the decision logic and corrupt-file cases. One gap worth closing: the distributed branch of update_and_check (dist.is_initialized() path — worker_ts sizing, all_gather_object, and broadcast_object_list) and the rank≠0 side of the RANK gate are never executed by these tests, and that's the code whose failure mode is a deadlock or rank desync in real multi-worker jobs. Both are cheap to cover without GPUs by mocking dist.is_initialized/get_world_size/all_gather_object/broadcast_object_list (e.g. an all_gather_object side_effect filling [1000.0, -1.0] and asserting rank 0 writes the quorum value; a broadcast_object_list side_effect setting the decision and asserting a RANK=1 process returns the broadcast result without touching the filesystem). Non-blocking suggestion.

[github-actions]

Review summary

Reviewed across code quality, performance, test coverage, documentation accuracy, and robustness. Overall this is a well-built feature: the hand-over decision is broadcast so all ranks exit the loops together (keeping the every-step collectives matched), the quorum math is reused from checkpoint_util.quorum_event_time, the progress write is atomic (tmp + os.replace), env config fails fast on every rank, and the doc was verified against the implementation with no discrepancies found (env names, 300 s default, cadence, checkpoint-skip and restart semantics all check out).

Findings (inline, all non-blocking):

1. Object collectives on the default NCCL group — all_gather_object / broadcast_object_list run on the default process group instead of a gloo side group like the sibling collectives in the same loop (checkpoint_util, online_dense_export_util), adding H2D/D2H copies and a device sync to the train loop.
2. OverflowError on a huge-int data_ts in the successor's progress.json — the one corrupt-file shape that escapes the OSError, ValueError handling and crashes rank 0 mid-collective instead of degrading to "no hand-over"; one-line fix.
3. Distributed branch has no test coverage — the dist path of update_and_check and the rank≠0 side are never executed by the tests; both are cheap to cover with mocked dist.

Two smaller observations not worth inline threads: rank 0 does two remote-filesystem round trips per check (write own progress + open a usually-absent successor file) while the other ranks block in the broadcast — fine at the default log_step_count_steps, worth remembering if the cadence is ever lowered; and a non-numeric ONLINE_LEARNING_HANDOVER_TOLERANCE_SECS value raises Python's generic could not convert string to float without naming the env var, unlike the crafted message for the negative/NaN cases.

🤖 Generated with Claude Code

changed needed

[eric-gecheng]

full_export 只需要导出到oss即可， 不需要将embedding 上传featureDB， 只有增量训练才需要上传featureDB

[eric-gecheng]

只要挂载一个oss 目录， 就是T 天的全量导出目录。
增量训练 stream_T会直接在此目录中创建dense_hot_export/current.json

[eric-gecheng]

无需指定current_json 字段。

[eric-gecheng]

processor 无需配置distributed_embedding， processor 会自动判断。

[eric-gecheng]

改为：

• feature_store_embedding_polling： 从FeatureDB 轮询增量embedding的配置。 必填项为feature_view_name和version， 且需与训练时写入的feature view和version一致。 选填参数sparse_poll_interval_secs， 不填时默认为 0，表示由processor按同频率轮询dense和sparse， 若设置>0, 则可以以更高的频率拉取embedding并更新。

[eric-gecheng]

改为：

• sparse_poll_interval_secs（位于 feature_store_embedding_polling 内）：默认 0，即不做周期扫描，只在加载新的 dense 版本同时拉取一次增量embedding。 拉取的频率由dense_hot_load.dense_poll_interval_minutes控制。 若配置sparse_poll_interval_secs>0, 则可以让embedding更快的生效。

[eric-gecheng]

如 INPUT_TILE， 这个可以删除， 因为input_tile 在distributed export 下默认就是 3。

[eric-gecheng]

"distributed_embedding": True,

这个无需配置

[eric-gecheng]

dense_hot_load 里可以增加配置这个配置作为示例：

"dense_poll_interval_minutes": 1,

[eric-gecheng]

增量训练的model_dir 应该配置成T 日的模型全量导出目录