添加 Family#

上一节仅生成 024 时效的 GRIB2 数据产品，本节中我们为 CMA-GFS 00 时次生成的 10 天 240 小时预报都生成 GRIB2 产品。

下面将为每个时效创建一个 Family 节点。

修改工作流定义#

更新 ${TUTORIAL_HOME}/def 中的工作流定义文件 cma_gfs_post.py：

 1import os
 2
 3import ecflow
 4
 5
 6def slurm_serial(class_name, wckey):
 7    variables = {
 8        "ECF_JOB_CMD": "slsubmit6 %ECF_JOB% %ECF_NAME% %ECF_TRIES% %ECF_TRYNO% %ECF_HOST% %ECF_PORT%",
 9        "ECF_KILL_CMD": "slcancel4 %ECF_RID% %ECF_NAME% %ECF_HOST% %ECF_PORT%",
10            "CLASS": class_name,
11        "WCKEY": wckey,
12    }
13    return variables
14
15
16def slurm_parallel(nodes, tasks_per_node, class_name, wckey):
17    variables = {
18        "ECF_JOB_CMD": "slsubmit6 %ECF_JOB% %ECF_NAME% %ECF_TRIES% %ECF_TRYNO% %ECF_HOST% %ECF_PORT%",
19        "ECF_KILL_CMD": "slcancel4 %ECF_RID% %ECF_NAME% %ECF_HOST% %ECF_PORT%",
20        "NODES": nodes,
21        "TASKS_PER_NODE": tasks_per_node,
22            "CLASS": class_name,
23        "WCKEY": wckey,
24    }
25    return variables
26
27
28current_path = os.path.dirname(__file__)
29tutorial_base = os.path.abspath(os.path.join(current_path, "../"))
30def_path = os.path.join(tutorial_base, "def")
31ecfout_path = os.path.join(tutorial_base, "ecfout")
32program_base_dir = os.path.join(tutorial_base, "program/cma-gfs-post-program")
33run_base_dir = os.path.join(tutorial_base, "workdir")
34
35defs = ecflow.Defs()
36
37with defs.add_suite("cma_gfs_post") as suite:
38    suite.add_variable("PROGRAM_BASE_DIR", program_base_dir)
39    suite.add_variable("RUN_BASE_DIR", run_base_dir)
40
41    suite.add_variable("ECF_INCLUDE", os.path.join(def_path, "include"))
42    suite.add_variable("ECF_FILES", os.path.join(def_path, "ecffiles"))
43
44    suite.add_variable("ECF_DATE", "20230806")
45    suite.add_variable("HH", "00")
46
47    suite.add_limit("total_tasks", 10)
48    suite.add_inlimit("total_tasks")
49
50    forecast_hour_list = [ f"{hour:03}" for hour in range(0, 241, 3)]
51
52    for forecast_hour in forecast_hour_list:
53        with suite.add_family(forecast_hour) as fm_hour:
54            fm_hour.add_variable("FHOUR", forecast_hour)
55
56            with fm_hour.add_task("pre_data2grib2") as tk_pre_data2grib2:
57                tk_pre_data2grib2.add_variable(slurm_serial("serial", "105-09"))
58
59            with fm_hour.add_task("data2grib2") as tk_data2grib2:
60                tk_data2grib2.add_variable(slurm_parallel(4, 64, "normal", "105-09"))
61                tk_data2grib2.add_trigger("./pre_data2grib2 == complete")
62
63print(defs)
64def_output_path = str(os.path.join(def_path, "cma_gfs_post.def"))
65defs.save_as_defs(def_output_path)

47-48 行创建一个 Limit，并将该 Limit 应用到 cma_gfs_post 节点上，限制同时运行的任务不能超过 10 个。

警告

对于需要运行大量任务的工作流，一定要限制同时运行的任务数，避免同一时间大量提交作业导致 ecFlow 服务卡死。

50-54 行为每个时效创建创建 Family，逐三小时创建以三位时效数字作为名称的任务。

更新工作流#

与前一节一样，需要重新生成 def 文件并更新到 ecFlow 服务中。 在更新之前，我们先在 ecFlow UI 中将 cma_gfs_post 节点挂起，避免更新后系统提交大量作业。

更新工作流：

cd ${TUTORIAL_HOME}/def
python3 cma_gfs_post.py
ecflow_client --port 43083 --replace /cma_gfs_post cma_gfs_post.def

更新任务脚本#

在 ${TUTORIAL_HOME}/def/ecffiles 目录下更新 ecf 脚本 pre_data2grib2.ecf 和 data2grib2.ecf。 使用 FHOUR 表示时效，将 forecast_hour=024 替换为 forecast_hour=%FHOUR%。

运行任务#

创建脚本后，可以恢复 cma_gfs_post 自动运行。 右键单击 cma_gfs_post，选择 Resume，恢复工作流运行。 可以看到前 10 个作业开始运行：

../_images/ecflow-ui-run-limit.png

因为我们创建的 Limit (total_tasks) 限制最多运行 10 个任务，所以有 1 个任务运行结束后第 11 个任务才会自动运行。