Просмотр исходного кода

fix(datax): job.setting.speed 只留 channel,恢复多通道并发

生成器把 byte/record 同时写进全局 job.setting.speed 和每通道 core.transport,
DataX 按「全局÷每通道=1」推导并发、且 byte/record 优先级压过 channel,实际永远 1 通道;
从 job.setting.speed 去掉 byte/record,channel(6/10)才生效。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
tianyu.chu 2 недель назад
Родитель
Сommit
a523562f03
1 измененных файлов с 5 добавлено и 3 удалено
  1. 5 3
      dw_base/datax/job_config_generator.py

+ 5 - 3
dw_base/datax/job_config_generator.py

@@ -16,7 +16,8 @@ class JobConfigGenerator(object):
     生成 DataX 作业配置文件(json)。
 
     speed 三参走三级合并:L1 conf/datax-tuning.conf < L2 ini [speed] 段 < L3 cli_speed_overrides。
-    合并后的 speed 写进 job.setting.speed + core.transport.channel.speed。
+    合并后:channel 写 job.setting.speed(DataX 按 channel 数并发);byte/record 只写
+    core.transport.channel.speed 当每通道限速——不可再塞进 job.setting.speed,否则并发被压成 1。
     """
 
     def __init__(self, base_dir: str, generator_config: str, start_date: str, stop_date: str, output: str,
@@ -51,10 +52,11 @@ class JobConfigGenerator(object):
         tuning_conf_path = os.path.join(self.base_dir, 'conf', 'datax-tuning.conf')
         l1 = load_tuning_conf(tuning_conf_path)
         merged = merge_speed(l1, self.config_parser, self.cli_speed_overrides)
+        # job.setting.speed 只放 channel:DataX 并发数按「全局 byte/record ÷ 每通道 byte/record」推导,
+        # 且 byte/record 限制优先级高于 channel。若这里也放 byte/record(与 core 每通道同值),
+        # 会算出 全局÷每通道=1 → 把 channel 压成 1 通道。byte/record 只作每通道限速放进 core.transport。
         speed = {
             JOB_SETTING_SPEED_CHANNEL: merged['channel'],
-            JOB_SETTING_SPEED_BYTE: merged['byte'],
-            JOB_SETTING_SPEED_RECORD: merged['record'],
         }
         core_speed = {
             'transport': {