EC Scheduler - Worker Configuration

### **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:

```bash
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:**

```python
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:**

```python
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.