Как условно запустить части вашего трубопровода Кедро?

Question

У меня большой конвейер, на запуск которого уходит несколько часов. Небольшая его часть должна запускаться довольно часто, как мне запустить ее, не запуская весь конвейер?

Answer 1

Есть несколько способов указать, какие узлы или части вашего конвейера запускать.

1. Использовать kedro run такие параметры как --to-nodes/--from-nodes/--node чтобы явно определить, что нужно запустить.

2. В kedro>=0.15.2 вы можете определить несколько конвейеров, а затем запустить только один из них с kedro run --pipeline <name>. Если нет--pipelineуказан параметр, запускается конвейер по умолчанию. Конвейер по умолчанию может объединять несколько других конвейеров. Дополнительная информация об использовании модульных конвейеров: https://kedro.readthedocs.io/en/latest/04_user_guide/06_pipelines.html

3. Используйте теги. Отметьте небольшую часть своего конвейера чем-то вроде "small", а затем выполнитеkedro run --tag small. Подробнее здесь: https://kedro.readthedocs.io/en/latest/04_user_guide/05_nodes.html

Answer 2

Я бы рекомендовал настроить правильную работу ваших тегов или конвейеров из cli, как предложил @idanov. В конечном итоге вам будет намного проще перейти на производство. Я бы также добавил, что вы можете немного обрезать конвейер и запускать его внутри Python, вот несколько примеров.

фильтровать по тегам

nodes = pipeline.only_nodes_with_tags('cars')

фильтр по узлу

nodes = pipeline.only_nodes('b_int_cars')

узлы фильтрации, такие как

query_string = 'cars'
nodes = [
   node.name 
   for node in pipeline.nodes 
   if query_string in node.name
   ]
pipeline.only_nodes(*nodes)

только узлы с тегами или

nodes = pipeline.only_nodes_with_tags('cars', 'trains')

только узлы с тегами и

raw_nodes = pipeline.only_nodes_with_tags('raw')
car_nodes = pipeline.only_nodes_with_tags('cars')
raw_car_nodes = raw_nodes & car_nodes

raw_nodes = (
   pipeline
   .only_nodes_with_tags('raw')
   .only_nodes_with_tags('cars')
   )

добавить трубопроводы

car_nodes = pipeline.only_nodes_with_tags('cars')
train_nodes = pipeline.only_nodes_with_tags('trains')
transportation_nodes = car_nodes + train_nodes

Выше был отрывок из моих личных заметок кедро.