代码之家 › 专栏 › 技术社区 › Midiparse

在气流中创建动态池

airflow

1

Midiparse · 技术社区 · 8 年前

我想我可以通过在创建集群后从任务动态创建一个池来解决这个问题,并在计算任务完成后删除它。我想我可以用模板 pool 参数,使计算任务使用此动态创建的群集。

# execute registers a pool and returns with the pool name
create_pool = CreatePoolOperator(
    slots=4,
    task_id='create_pool',
    dag=self
)

# the pool parameter is templated
computation = ComputeOperator(
    task_id=compute_subtask_name,
    pool="{{ ti.xcom_pull(task_ids='create_pool') }}",
    dag=self
)

create_pool >> computation

但是这样计算任务永远不会被触发。所以我认为pool参数在被模板化之前保存在任务实例中。我想听听你对如何达到理想行为的想法。

1 回复 | 直到 8 年前

1

0

joebeeson 8 年前

与其尝试让动态池工作,不如看看 concurrency 属性打开 airflow.models.DAG 会成功的。它将限制当前进程运行中正在运行的任务的数量。

2

0

Kyle Bridenstine 7 年前

您希望能够在DAG中集中任务,但仅限于该特定DAG运行。所以试着在你的任务上创建一个自定义池。我脑子里有一些伪代码

List<String> tasksPoolQueue = new ArrayList<String>();

def taskOnesFunction() 

  while true:

    if tasksPoolQueue.get(0) == "taskOnesTurn":
       print("Do some work it's your turn")

       # Delete this run from the list and shift the list over to the left one index
       # So that the next value is now the first value in the list
       tasksPoolQueue.delete(0)

       return 0

    else:
      sleep(10 seconds)

def taskTwosFunction()

  while true:

    if tasksPoolQueue.get(0) == "taskTwosTurn":
       print("Do some work it's your turn")

       # Delete this run from the list and shift the list over to the left one index
       # So that the next value is now the first value in the list
       tasksPoolQueue.delete(0)

       return 0

    else:
      sleep(10 seconds)

def createLogicalOrderingOfTaskPoolQueue():

    if foobar == true:
      tasksPoolQueue[0] = "taskOnesTurn"
      tasksPoolQueue[1] = "taskTwosTurn"
    else:
      tasksPoolQueue[0] = "taskTwosTurn"
      tasksPoolQueue[1] = "taskOnesTurn"

    return 0


determine_pool_queue_ordering = PythonOperator(
    task_id='determine_pool_queue_ordering',
    retries=0,
    dag=dag,
    provide_context=True,
    python_callable=createLogicalOrderingOfTaskPoolQueue,
    op_args=[])

task1 = PythonOperator(
    task_id='task1',
    retries=0,
    dag=dag,
    provide_context=True,
    python_callable=taskOnesFunction,
    op_args=[])

task2= PythonOperator(
    task_id='task2',
    retries=0,
    dag=dag,
    provide_context=True,
    python_callable=taskTwosFunction,
    op_args=[])

determine_pool_queue_ordering.set_downstream(task1)
determine_pool_queue_ordering.set_downstream(task2)

所以希望每个人都能遵循我的伪代码。我不知道创建自定义池的最佳方法是不引入“竞争条件”,所以这个列表队列的想法是我第一眼看到的。但这里的要点是task1和task2将同时运行,但在它们的函数内部,我可以让它做一些有意义的事情,直到它超过if语句,阻止它运行真正的代码。

第一个任务将使用列表动态设置哪些任务首先运行以及以什么顺序运行。然后让所有需要在此自定义池中的函数引用该列表。因为我们的if语句只有在taskName在列表中第一个时才等于true,这实际上意味着一次只能运行一个任务。列表中的第一个任务在处理完需要执行的任何操作后,将从列表中删除自己。然后,其他任务将在等待任务名称出现在列表中的第一个时休眠。

所以只要做一些类似我的自定义逻辑。

3

0

Dave C 7 年前

这里有一个操作符,如果池不存在,它会创建一个池。

from airflow.api.common.experimental.pool import get_pool, create_pool
from airflow.exceptions import PoolNotFound
from airflow.models import BaseOperator
from airflow.utils import apply_defaults


class CreatePoolOperator(BaseOperator):
    # its pool blue, get it?
    ui_color = '#b8e9ee'

    @apply_defaults
    def __init__(
            self,
            name,
            slots,
            description='',
            *args, **kwargs):
        super(CreatePoolOperator, self).__init__(*args, **kwargs)
        self.description = description
        self.slots = slots
        self.name = name

    def execute(self, context):
        try:
            pool = get_pool(name=self.name)
            if pool:
                self.log(f'Pool exists: {pool}')
                return
        except PoolNotFound:
            # create the pool
            pool = create_pool(name=self.name, slots=self.slots, description=self.description)
            self.log(f'Created pool: {pool}')

可以用类似的方式删除池。