EC Scheduler - Worker Configuration

[irfanghat]

EC Scheduler - Worker Configuration

1. Job Assignment:

   • When a job is submitted, the scheduler checks how many threads the worker has available (based on system resources).

   • The scheduler assigns each job to a specific worker (e.g., job_id1 -> worker_1, job_id2 -> worker_2), based on available resources.

2. Thread Allocation:

   • dbt can be configured to run multiple models in parallel using the --threads flag. For instance, if worker_1 has 4 threads available, the scheduler could assign dbt run --select my_model --threads 4 to worker_1.

   • This means each worker is effectively running multiple models in parallel, depending on the number of threads available.

3. Scaling Workers:

   • As more jobs are submitted, new workers are created (or existing ones are reused) to run the jobs.

   • This setup makes the system easy to scale — simply adding more workers or adjusting the number of threads available for each worker based on system load.

---

How to Implement This Logic

Let’s break it down further into an example flow:

Scheduler:

1. The scheduler receives a job, e.g., dbt run --select my_model.

2. It checks the available threads on the worker node. Let’s say worker_1 has 4 threads available.

3. It assigns dbt run --select my_model --threads 4 to worker_1.

4. The job is added to Redis, and the job ID is linked to the specific worker.

5. The scheduler tracks the job status in Redis or a database.

Worker:

1. Workers listen for jobs using ZeroMQ or Redis.

2. Each worker receives a job and checks if it has enough threads to handle it.

3. The worker executes the dbt command using the --threads flag for parallel execution:

   dbt run --select my_model --threads 4

4. Once the job is complete, the worker reports back to the scheduler with the job’s result (success/failure).

Redis (for Job Queueing and Status Tracking):

• The scheduler can use Redis to store job statuses (e.g., pending, running, completed).

• Redis also helps ensure jobs aren’t lost, and failed jobs can be retried.

---

Example Code Flow

Scheduler:

import zmq
import redis
import json
import os

context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://*:5555")  # Publish jobs to workers

r = redis.Redis(host='localhost', port=6379, db=0)

# Check available threads for each worker (just for the example)
def get_available_threads(worker_id):
    # Logic to check system resources and assign threads
    # Example: return a fixed number for now
    if worker_id == 1:
        return 4  # Worker 1 has 4 threads available
    elif worker_id == 2:
        return 2  # Worker 2 has 2 threads available
    else:
        return 1  # Default to 1 thread for other workers

def submit_job(job, worker_id):
    threads = get_available_threads(worker_id)
    job_id = f"job_{job['model']}"

    # Add job to Redis queue
    r.lpush('job_queue', json.dumps(job))
    
    # Dispatch job to worker via ZeroMQ
    socket.send_string(f"New Job: {job_id} - {job['model']} - {threads} threads")
    print(f"Job {job_id} assigned to worker_{worker_id} with {threads} threads.")

# Example job submission
job = {
    'model': 'my_model',
    'command': 'dbt run --select my_model',
}
submit_job(job, worker_id=1)

Worker:

import zmq
import redis
import json
import subprocess

context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://localhost:5555")
socket.setsockopt_string(zmq.SUBSCRIBE, "")  # Subscribe to all messages

r = redis.Redis(host='localhost', port=6379, db=0)

def process_job(job):
    print(f"Processing job: {job['model']} with {job['threads']} threads")
    result = subprocess.run(job['command'], shell=True, capture_output=True)
    return result.stdout.decode() if result.returncode == 0 else result.stderr.decode()

while True:
    # Receive job via ZeroMQ
    message = socket.recv_string()
    print(f"Received job message: {message}")
    
    # Fetch the job from Redis queue
    job_data = r.brpop('job_queue')  # Blocking pop from Redis queue
    job = json.loads(job_data[1])    # Parse job

    # Execute the dbt command with the assigned number of threads
    job['command'] = f"dbt run --select {job['model']} --threads {job['threads']}"
    output = process_job(job)
    
    # Handle job result (send results back to scheduler, log, etc.)
    print(f"Job result: {output}")

---

Enhancing This Setup

• Job Dependencies: For complex workflows (e.g., DBT models with dependencies), you can check if one job depends on another and ensure that jobs are executed in the correct order.

• Fault Tolerance: Implement retries in case of failures. If a worker fails to execute a job, it can either retry or report the failure, allowing the scheduler to assign the job to another worker.

• Monitoring: Add monitoring to check job progress, handle failed jobs, and ensure that resources are being used efficiently.

---

Benefits of This Approach:

1. Efficient Resource Usage: Workers use the available threads on their nodes, making sure resources are allocated efficiently.

2. Simple Scaling: New workers can be added easily as demand grows. Each worker can handle jobs independently, with dbt’s built-in thread management scaling execution.

3. Parallel Execution: By leveraging dbt’s ability to run models in parallel, workers can handle multiple models concurrently, improving throughput.