如何实现Notebook触发方式判断:自动(定时任务)或手动执行?
在Notebook中区分手动/自动运行并切换日期参数的实现方案
当然可以实现这个逻辑,下面分几种常用Notebook环境给出具体方案:
通用思路核心
通过识别运行上下文的标识(环境变量、运行ID、参数传入情况)来判断运行模式,分支处理日期参数:
- 自动运行(定时任务):使用当前系统日期
- 手动触发:使用用户输入/指定的日期
1. Jupyter Notebook/Lab 环境
方案:用环境变量标记运行模式
定时任务(如cron、Airflow调用)启动Notebook时,先设置环境变量,代码中读取该变量判断模式:
import os from datetime import datetime # 读取环境变量,默认手动模式 run_mode = os.getenv("NOTEBOOK_RUN_MODE", "manual") if run_mode == "auto": # 自动运行:用当前日期 target_date = datetime.today().strftime("%Y-%m-%d") print(f"自动运行模式,使用当前日期:{target_date}") else: # 手动运行:获取用户输入 target_date = input("请输入目标日期(格式YYYY-MM-DD):").strip() # 可选:添加日期格式校验 print(f"手动运行模式,使用指定日期:{target_date}")
定时任务启动命令示例(cron中):
NOTEBOOK_RUN_MODE=auto jupyter nbconvert --to notebook --execute your_notebook.ipynb
2. Databricks Notebook 环境
Databricks作业调度会自动生成唯一的run_id,手动交互式运行时无此标识,可通过API判断:
from datetime import datetime from pyspark.sql import SparkSession spark = SparkSession.builder.getOrCreate() # 获取运行上下文ID try: run_id = spark.conf.get("spark.databricks.job.id") # 存在job_id则为自动调度运行 target_date = datetime.today().strftime("%Y-%m-%d") print(f"自动调度运行,使用当前日期:{target_date}") except Exception: # 无job_id则为手动运行 target_date = input("请输入目标日期(格式YYYY-MM-DD):").strip() print(f"手动运行模式,使用指定日期:{target_date}")
或者用dbutils工具直接判断:
from datetime import datetime run_context = dbutils.notebook.entry_point.getDbutils().notebook().getContext() run_id = run_context.runId().getOrElse(None) if run_id and "job" in run_id: target_date = datetime.today().strftime("%Y-%m-%d") print(f"自动作业运行,使用当前日期:{target_date}") else: target_date = input("请输入目标日期(格式YYYY-MM-DD):").strip() print(f"手动运行模式,使用指定日期:{target_date}")
3. 通用参数传递方案(跨环境)
通过命令行参数传入日期,自动运行时不传(代码默认用当前日期),手动运行时指定参数:
import argparse from datetime import datetime parser = argparse.ArgumentParser(description="Notebook日期参数设置") parser.add_argument("--target-date", type=str, help="目标日期(格式YYYY-MM-DD),手动运行时指定") args = parser.parse_args() if args.target_date: # 手动运行:使用传入的日期 target_date = args.target_date print(f"手动运行模式,使用指定日期:{target_date}") else: # 自动运行:使用当前日期 target_date = datetime.today().strftime("%Y-%m-%d") print(f"自动运行模式,使用当前日期:{target_date}")
手动运行时启动命令:
jupyter nbconvert --to notebook --execute your_notebook.ipynb --ExecutePreprocessor.timeout=3600 -- --target-date "2024-05-20"
自动定时任务启动时,直接运行不带--target-date参数的命令即可。
内容的提问来源于stack exchange,提问作者adahero
相关产品推荐
相关产品推荐

