Yes, this looks like a good implementation! You're using several good practices here, including: 1. Using

.map()

for concurrent execution 2. Proper use of

wait()

to collect futures 3. Good state checking with

is_completed()

4. Retries on both tasks 5. Using DaskTaskRunner for distributed execution 6. Proper result tracking with success/failure counts A few small suggestions to make it even better:

python
from prefect import flow, task, get_run_logger
import random 
from prefect.futures import wait
from dask_jobqueue import PBSCluster
from prefect_dask import DaskTaskRunner
from typing import List, Any

@task(retries=3)
def task1_logic(url: Any) -> float:
    pval = random.random() 
    if pval < 0.5:  # Note: using < instead of <
        raise ValueError("some failure in step 1")
    return pval 
    
@task(retries=4)
def task2_logic(pval: float) -> str:
    pval2 = random.random()
    if pval2 < 0.5:
        raise ValueError("some failure in step 2")
    return "ok"

@flow(task_runner=DaskTaskRunner(),
      persist_result=True)
def process_urls(urls: List[Any] = None) -> dict:
    if urls is None:
        urls = list(range(10))
        
    logger = get_run_logger()
    
    # Submit all tasks
    task1_futs = task1_logic.map(urls)
    task2_futs = task2_logic.map(task1_futs)
    
    # Wait for completion
    done, not_done = wait(task2_futs)
    
    # Track results
    successes = 0
    failures = 0
    failed_tasks = []
    
    for fut in done:
        if fut.state.is_completed():
            <http://logger.info|logger.info>(f"Task completed successfully: {fut.result()}")
            successes += 1
        else:
            logger.error(f"Task failed: {fut.task_run_id} - State: {fut.state}")
            failed_tasks.append({
                "task_run_id": fut.task_run_id,
                "state": str(fut.state),
                "error": str(fut.state.result()) if fut.state.result() else None
            })
            failures += 1

    return {
        "total_tasks": len(done) + len(not_done),
        "completed": len(done),
        "not_done": len(not_done),
        "successes": successes,
        "failures": failures,
        "failed_tasks": failed_tasks
    }

if __name__ == "__main__":
    results = process_urls()
    print(f"Execution Summary:")
    print(f"Total Tasks: {results['total_tasks']}")
    print(f"Completed: {results['completed']}")
    print(f"Not Done: {results['not_done']}")
    print(f"Successes: {results['successes']}")
    print(f"Failures: {results['failures']}")
    if results['failed_tasks']:
        print("\nFailed Tasks:")
        for task in results['failed_tasks']:
            print(f"- Task {task['task_run_id']}: {task['error']}")

Key improvements: 1. Added type hints for better code clarity 2. Made the URLs parameter configurable 3. Enhanced error tracking to capture specific failure information 4. Using logger instead of print statements 5. Structured return value with complete execution summary 6. Better error reporting in the main execution block 7. Fixed HTML entity (

<

to

<

) The code will now give you more detailed information about failures while maintaining all the concurrent execution benefits of your original implementation. The structured return value makes it easier to process the results programmatically if needed. Also, since you're using DaskTaskRunner, make sure your PBS cluster configuration matches your needs. You might want to add cluster configuration if you haven't already: ```python @flow(task_runner=DaskTaskRunner( cluster_

This issue typically occurs when the PBS cluster isn't properly initialized or can't schedule the jobs. Let's fix the PBS cluster configuration:

python
from prefect import flow, task, get_run_logger
import random 
from prefect.futures import wait
from dask_jobqueue import PBSCluster
from prefect_dask import DaskTaskRunner
from distributed import Client
import os

# Define PBS cluster configuration
pbs_cluster_kwargs = {
    "name": "dask-worker",
    "queue": "your_queue_name",  # Replace with your PBS queue name
    "cores": 1,
    "memory": "4GB",
    "processes": 1,
    "walltime": "01:00:00",
    "resource_spec": "select=1:ncpus=1:mem=4gb",
    # Add any necessary PBS directives
    "job_extra": [
        "-l select=1:ncpus=1:mem=4gb",
        "-l walltime=01:00:00",
    ],
    "local_directory": os.getenv("TMPDIR", "/tmp"),  # Use TMPDIR if available
}

# Adaptation settings
adapt_kwargs = {
    "minimum": 1,
    "maximum": 20,
    "interval": "30s"
}

@task(retries=3)
def task1_logic(url):
    pval = random.random() 
    if pval < 0.5:
        raise ValueError("some failure in step 1")
    return pval 
    
@task(retries=4)
def task2_logic(pval):
    pval2 = random.random()
    if pval2 < 0.5:
        raise ValueError("some failure in step 2")
    return "ok"

@flow(task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs=adapt_kwargs
      ),
      persist_result=True)
def process_urls():
    logger = get_run_logger()
    
    # Log cluster information
    client = Client.current()
    <http://logger.info|logger.info>(f"Dask dashboard: {client.dashboard_link}")
    <http://logger.info|logger.info>(f"Cluster status: {client.status}")
    
    urls = list(range(10))
    
    # Submit tasks
    task1_futs = task1_logic.map(urls)
    
    # Wait a bit for workers to come online if needed
    client.wait_for_workers(1)
    
    task2_futs = task2_logic.map(task1_futs)
    done, not_done = wait(task2_futs)
    
    successes = 0
    failures = 0
    for fut in done:
        if fut.state.is_completed():
            <http://logger.info|logger.info>(f"Task completed: {fut.result()}")
            successes += 1
        else:
            logger.error(f"Task failed: {fut.task_run_id} - State: {fut.state}")
            failures += 1
    
    <http://logger.info|logger.info>(f"Done: {len(done)}")
    <http://logger.info|logger.info>(f"Not Done: {len(not_done)}")
    <http://logger.info|logger.info>(f"Failed: {failures}")
    <http://logger.info|logger.info>(f"Successes: {successes}")
    
    # Log final cluster status
    <http://logger.info|logger.info>(f"Final cluster status: {client.status}")

if __name__ == "__main__":
    process_urls()

Key changes to fix the Pending state issue: 1. Proper PBS Configuration: Make sure to adjust these values according to your cluster: - `queue`: Your PBS queue name - `resource_spec`: Match your cluster's resource specification format - `job_extra`: Include any necessary PBS directives - Adjust memory and cores based on your needs 2. Adaptation Settings: More detailed settings for worker scaling: ```python adapt_kwargs = { "minimum": 1, # At least one worker "maximum": 20, # Maximum number of workers "interval": "30s" # How often to check for scaling