kedro/runner/parallel_runner.py
def _run(  # pylint: disable=too-many-locals,useless-suppression
        self, pipeline: Pipeline, catalog: DataCatalog, run_id: str = None
    ) -> None:
        """The abstract interface for running pipelines.

        Args:
            pipeline: The ``Pipeline`` to run.
            catalog: The ``DataCatalog`` from which to fetch data.
            run_id: The id of the run.

        Raises:
            AttributeError: when the provided pipeline is not suitable for
                parallel execution.
            Exception: in case of any downstream node failure.

        """
        nodes = pipeline.nodes
        self._validate_catalog(catalog, pipeline)
        self._validate_nodes(nodes)

        load_counts = Counter(chain.from_iterable(n.inputs for n in nodes))
        node_dependencies = pipeline.node_dependencies
        todo_nodes = set(node_dependencies.keys())
        done_nodes = set()  # type: Set[Node]
        futures = set()
        done = None
        max_workers = self._get_required_workers_count(pipeline)

        with ProcessPoolExecutor(max_workers=max_workers) as pool:
            while True:
                ready = {n for n in todo_nodes if node_dependencies[n] <= done_nodes}
                todo_nodes -= ready
                for node in ready:
                    futures.add(
                        pool.submit(
                            _run_node_synchronization,
                            node,
                            catalog,
                            self._is_async,
                            run_id,
                        )
                    )
                if not futures:
                    assert not todo_nodes, (todo_nodes, done_nodes, ready, done)
                    break
                done, futures = wait(futures, return_when=FIRST_COMPLETED)
                for future in done:
                    try:
                        node = future.result()
                    except Exception:
                        self._suggest_resume_scenario(pipeline, done_nodes)
                        raise
                    done_nodes.add(node)

                    # decrement load counts and release any data sets we've finished with
                    # this is particularly important for the shared datasets we create above
                    for data_set in node.inputs:
                        load_counts[data_set] -= 1
                        if (
                            load_counts[data_set] < 1
                            and data_set not in pipeline.inputs()
                        ):
                            catalog.release(data_set)
                    for data_set in node.outputs:
                        if (
                            load_counts[data_set] < 1
                            and data_set not in pipeline.outputs()
                        ):
                            catalog.release(data_set)
The abstract interface for running pipelines.

        Args:
            pipeline: The ``Pipeline`` to run.
            catalog: The ``DataCatalog`` from which to fetch data.
            run_id: The id of the run.

        Raises:
            AttributeError: when the provided pipeline is not suitable for
                parallel execution.
            Exception: in case of any downstream node failure.

        
The abstract interface for running pipelines.
The abstract interface for running pipelines .
def _run(  
        self, pipeline: Pipeline, catalog: DataCatalog, run_id: str = None
    ) -> None:
        nodes = pipeline.nodes
        self._validate_catalog(catalog, pipeline)
        self._validate_nodes(nodes)
        load_counts = Counter(chain.from_iterable(n.inputs for n in nodes))
        node_dependencies = pipeline.node_dependencies
        todo_nodes = set(node_dependencies.keys())
        done_nodes = set()  
        futures = set()
        done = None
        max_workers = self._get_required_workers_count(pipeline)
        with ProcessPoolExecutor(max_workers=max_workers) as pool:
            while True:
                ready = {n for n in todo_nodes if node_dependencies[n] <= done_nodes}
                todo_nodes -= ready
                for node in ready:
                    futures.add(
                        pool.submit(
                            _run_node_synchronization,
                            node,
                            catalog,
                            self._is_async,
                            run_id,
                        )
                    )
                if not futures:
                    assert not todo_nodes, (todo_nodes, done_nodes, ready, done)
                    break
                done, futures = wait(futures, return_when=FIRST_COMPLETED)
                for future in done:
                    try:
                        node = future.result()
                    except Exception:
                        self._suggest_resume_scenario(pipeline, done_nodes)
                        raise
                    done_nodes.add(node)
                    for data_set in node.inputs:
                        load_counts[data_set] -= 1
                        if (
                            load_counts[data_set] < 1
                            and data_set not in pipeline.inputs()
                        ):
                            catalog.release(data_set)
                    for data_set in node.outputs:
                        if (
                            load_counts[data_set] < 1
                            and data_set not in pipeline.outputs()
                        ):
                            catalog.release(data_set)
def _run ( self , pipeline : Pipeline , catalog : DataCatalog , run_id : str = None ) -> None : nodes = pipeline . nodes self . _validate_catalog ( catalog , pipeline ) self . _validate_nodes ( nodes ) load_counts = Counter ( chain . from_iterable ( n . inputs for n in nodes ) ) node_dependencies = pipeline . node_dependencies todo_nodes = set ( node_dependencies . keys ( ) ) done_nodes = set ( ) futures = set ( ) done = None max_workers = self . _get_required_workers_count ( pipeline ) with ProcessPoolExecutor ( max_workers = max_workers ) as pool : while True : ready = { n for n in todo_nodes if node_dependencies [ n ] <= done_nodes } todo_nodes -= ready for node in ready : futures . add ( pool . submit ( _run_node_synchronization , node , catalog , self . _is_async , run_id , ) ) if not futures : assert not todo_nodes , ( todo_nodes , done_nodes , ready , done ) break done , futures = wait ( futures , return_when = FIRST_COMPLETED ) for future in done : try : node = future . result ( ) except Exception : self . _suggest_resume_scenario ( pipeline , done_nodes ) raise done_nodes . add ( node ) for data_set in node . inputs : load_counts [ data_set ] -= 1 if ( load_counts [ data_set ] < 1 and data_set not in pipeline . inputs ( ) ) : catalog . release ( data_set ) for data_set in node . outputs : if ( load_counts [ data_set ] < 1 and data_set not in pipeline . outputs ( ) ) : catalog . release ( data_set )
The abstract interface for running pipelines.
The abstract interface for running pipelines .
# pylint: disable=too-many-locals,useless-suppression """The abstract interface for running pipelines.

        Args:
            pipeline: The ``Pipeline`` to run.
            catalog: The ``DataCatalog`` from which to fetch data.
            run_id: The id of the run.

        Raises:
            AttributeError: when the provided pipeline is not suitable for
                parallel execution.
            Exception: in case of any downstream node failure.

        """ # type: Set[Node] # decrement load counts and release any data sets we've finished with # this is particularly important for the shared datasets we create above
when the provided pipeline is not suitable for
parallel execution.
when the provided pipeline is not suitable for parallel execution .
AttributeError
in case of any downstream node failure.
in case of any downstream node failure .
Exception
self
pipeline
Pipeline
The ``Pipeline`` to run.
The ` ` Pipeline ` ` to run .
catalog
DataCatalog
The ``DataCatalog`` from which to fetch data.
The ` ` DataCatalog ` ` from which to fetch data .
run_id
str
The id of the run.
The id of the run .