气流：如何在不同时间安排匕首？

如果希望每天随机调度一次 dag，请编写 python 帮助程序代码。 在定义 dag 之前，在 dag 代码中，放置一个种子随机（不会随日期变化）以创建伪兰特。

在此示例中，我已将完整日期转换为数字日期，但您可以使用您喜欢的任何方法。

像下面的代码应该可以工作。

random.seed(int(datetime.date.today().strftime('%Y%m%d')))    
randomCronString="* {} * * *".format(random.randint(0,24))
dag=Dag('TestDag,schedule_interval=randomCronString,default_args=args, catchup=False)

编辑

Airflow 每 5 秒运行一次 dag，但通过使用种子随机，您强制随机仅在种子发生变化时更改（在这种情况下，当一天翻转时），但请记住，在大多数系统中，气流都在 UTC 上。

如果一遍又一遍地运行以下代码块

import random
import datetime

random.seed(int(datetime.date.today().strftime('%Y%m%d')))    
randomCronString="* {} * * *".format(random.randint(0,24))
print(randomCronString)

您将获得以下结果

random.seed(int(datetime.date.today().strftime('%Y%m%d')))    
randomCronString="* {} * * *".format(random.randint(0,24))

print(randomCronString)
* 1 * * *

random.seed(int(datetime.date.today().strftime('%Y%m%d')))    
randomCronString="* {} * * *".format(random.randint(0,24))
print(randomCronString)
* 1 * * *

random.seed(int(datetime.date.today().strftime('%Y%m%d')))    
randomCronString="* {} * * *".format(random.randint(0,24))
print(randomCronString)
* 1 * * *

random.seed(int(datetime.date.today().strftime('%Y%m%d')))    
randomCronString="* {} * * *".format(random.randint(0,24))
print(randomCronString)
* 1 * * *

random.seed(int(datetime.date.today().strftime('%Y%m%d')))    
randomCronString="* {} * * *".format(random.randint(0,24))
print(randomCronString)
* 1 * * *

random.seed(int(datetime.date.today().strftime('%Y%m%d')))    
randomCronString="* {} * * *".format(random.randint(0,24))
print(randomCronString)
* 1 * * *

random.seed(int(datetime.date.today().strftime('%Y%m%d')))    
randomCronString="* {} * * *".format(random.randint(0,24))
print(randomCronString)
* 1 * * *

random.seed(int(datetime.date.today().strftime('%Y%m%d')))    
randomCronString="* {} * * *".format(random.randint(0,24))
print(randomCronString)
* 1 * * *

random.seed(int(datetime.date.today().strftime('%Y%m%d')))    
randomCronString="* {} * * *".format(random.randint(0,24))
print(randomCronString)
* 1 * * *

random.seed(int(datetime.date.today().strftime('%Y%m%d')))    
randomCronString="* {} * * *".format(random.randint(0,24))
print(randomCronString)
* 1 * * *

random.seed(int(datetime.date.today().strftime('%Y%m%d')))    
randomCronString="* {} * * *".format(random.randint(0,24))
print(randomCronString)
* 1 * * *

添加一个延迟步骤（具有随机的睡眠间隔），并在实际做某事之前安排它怎么样？

例如

def _delay_dag():
    import random
    import time
    delay_by = random.randint(0,60*60*6) #e.g. to delay up to 6h from schedule time
    print(f'Delaying start by {delay_by} seconds..')
    time.sleep(delay_by)
    return True

def _do_something(ti, *kwargs):
    pass
    
with DAG('my_DAG_with_random_start', 
         schedule_interval='@daily', 
         tags=['test','random_start'],
         default_args=default_args) as dag:
    
    delay_dag = PythonOperator(
        task_id = 'delay_dag',
        python_callable=_delay_dag
    )

    do_something = PythonOperator(
        task_id = 'do_something',
        provide_context=True,
        python_callable=_do_something
    )
    
    delay_dag >> do_something