详解Python如何轻松实现定时执行任务

下面这个截图，就是使用 schedule 定时执行 Notebook 的例子

import schedule
import time
import papermill as pmdef job():
   pm.execute_notebook(
      'input.ipynb',
       'logs/oo.ipynb'
   )
schedule.every(10).seconds.do(job)
while True:
   schedule.run_pending()
   time.sleep(1)

所需要的包就这么几个（ipywidgets 是为了显示进度条，就不多做介绍了），

pip install schedule papermill ipywidgets

下面分别介绍下各个包的用途。

schedule

从官方提供的例子可以看出，只要会基本的英语就能明白如何方便定时了

import schedule
import time

# 要执行的任务
def job():
   print("I'm working...")

#每隔10s执行job
schedule.every(10).seconds.do(job)
#每隔10分钟执行job
schedule.every(10).minutes.do(job)
#每小时执行job
schedule.every().hour.do(job)
#在每天10:30分执行job
schedule.every().day.at("10:30").do(job)
#每5到10分钟（随机数）执行job
schedule.every(5).to(10).minutes.do(job)
#每星期一执行job
schedule.every().monday.do(job)
#每星期三13:15执行job
schedule.every().wednesday.at("13:15").do(job)
#每天欧洲阿姆斯特丹时区的12:42执行job
schedule.every().day.at("12:42", "Europe/Amsterdam").do(job)
#在每分钟的第17秒开始执行job
schedule.every().minute.at(":17").do(job)

def job_with_argument(name):
   print(f"I am {name}")
#每10秒钟执行一次job_with_argument（带参数）
schedule.every(10).seconds.do(job_with_argument, name="Peter")

while True:
   schedule.run_pending()
   time.sleep(1)

更多应用可以看官方文档[1]。

如果你只是在 Python 代码里定时一个或者多个工作，你并不需要 papermill，使用 papermill，是为了可以方便地定时执行你的 Notebook。

papermill

papermill[2]是一个用于参数化、执行和分析 Jupyter Notebooks 的工具。

也许您有一份财务报告，您希望在一个月的第一天或最后一天或一年的开始或结束时使用不同的值运行，使用参数可以使这项任务更容易。你想运行一个 Notebook 并根据它的结果，选择一个特定的 Notebook 来运行吗？您现在可以以编程方式 执行工作流程，而无需手动从笔记本复制和粘贴到笔记本。

我们即可以像上面的例子那样以 API 的方式使用，也可以以命令行的方式来运行，例如， 运行本地input.ipynb，参数分别为alpha和l1_ratio把结果保存到 AWS 的 s3 上，

papermill local/input.ipynb s3://bkt/output.ipynb -p alpha 0.6 -p l1_ratio 0.1

使用下面这段代码，我们可以把 notebook 中 print 输出转移输出到日志里，

from datetime import datetime
from datetime import timedelta
import logging, sys,re
#日志的名字以时间命名，方便查找
_date_start = (datetime.now() +timedelta(hours=0)).strftime('％Y-％m-％d ％H:％M:％S')
xxx = re.sub(r'[-\ \:]','',_date_start)
filename=f'logs/{xxx}.log'
#日志的基本设置
logging.basicConfig(filename=filename, level=logging.DEBUG
                   , format="％(asctime)s - ％(name)s - ％(levelname)s - ％(message)s")
logger = logging.getLogger()
# 将print转移输出到日志中
sys.stderr.write = logger.error
sys.stdout.write = logger.info

参考资料

[1]官方文档: https://schedule.readthedocs.io/

[2]papermill: https://github.com/nteract/papermill

来源：https://blog.csdn.net/qq_34160248/article/details/127480781