0
# Dagster Celery K8s
1
2
A Dagster integration that provides a Kubernetes-native distributed execution system by combining Celery task queuing with Kubernetes job orchestration. It enables scalable data pipeline execution through a two-tier architecture where a run worker Kubernetes Job traverses the execution plan and submits individual steps to Celery queues, while Celery workers spawn separate Kubernetes Jobs for each step execution.
3
4
## Package Information
5
6
- **Package Name**: dagster-celery-k8s
7
- **Language**: Python
8
- **Installation**: `pip install dagster-celery-k8s`
9
- **Version**: 0.27.9
10
11
## Core Imports
12
13
```python
14
from dagster_celery_k8s import CeleryK8sRunLauncher, celery_k8s_job_executor
15
```
16
17
Import version information:
18
19
```python
20
from dagster_celery_k8s import __version__
21
```
22
23
Import types for full type annotations:
24
25
```python
26
from typing import Optional, List, Dict, Any
27
from dagster import DagsterRun, LaunchRunContext
28
from dagster._core.launcher.base import CheckRunHealthResult
29
from dagster._serdes import ConfigurableClassData
30
from dagster_k8s import DagsterK8sJobConfig
31
```
32
33
## Basic Usage
34
35
```python
36
from dagster import job
37
from dagster_celery_k8s import celery_k8s_job_executor
38
39
# Define a job using the Celery K8s executor
40
@job(executor_def=celery_k8s_job_executor)
41
def my_distributed_job():
42
# Your job operations here
43
pass
44
```
45
46
Configure in Dagster instance (dagster.yaml):
47
48
```yaml
49
run_launcher:
50
module: dagster_k8s.launcher
51
class: CeleryK8sRunLauncher
52
config:
53
instance_config_map: "dagster-k8s-instance-config-map"
54
dagster_home: "/opt/dagster/dagster_home"
55
postgres_password_secret: "dagster-k8s-pg-password"
56
broker: "pyamqp://guest@localhost//"
57
backend: "rpc://"
58
59
execution:
60
config:
61
job_image: 'my_repo.com/image_name:latest'
62
job_namespace: 'dagster-execution'
63
broker: 'pyamqp://guest@localhost//'
64
backend: 'rpc://'
65
```
66
67
## Architecture
68
69
The dagster-celery-k8s package implements a two-tier distributed execution architecture:
70
71
1. **Run Worker Tier**: A Kubernetes Job that traverses the Dagster execution plan and submits individual steps to Celery queues
72
2. **Step Execution Tier**: Celery workers that pick up tasks and spawn separate Kubernetes Jobs for each step execution
73
74
This design provides:
75
- **Scalability**: Dynamic resource allocation through Kubernetes
76
- **Reliability**: Container orchestration with Kubernetes job management
77
- **Flexibility**: Celery's distributed task processing capabilities
78
- **Cloud-native**: Optimized for cloud environments with automatic resource management
79
80
## Capabilities
81
82
### Run Launcher
83
84
The primary component for launching Dagster runs as Kubernetes jobs with Celery task distribution.
85
86
```python { .api }
87
class CeleryK8sRunLauncher(RunLauncher, ConfigurableClass):
88
"""
89
Run launcher for Kubernetes-based execution with Celery task queuing.
90
91
Launches dagster runs as Kubernetes Jobs that traverse execution plans
92
and submit steps to Celery queues for distributed execution.
93
"""
94
95
def __init__(
96
self,
97
instance_config_map,
98
dagster_home,
99
postgres_password_secret,
100
load_incluster_config=True,
101
kubeconfig_file=None,
102
broker=None,
103
backend=None,
104
include=None,
105
config_source=None,
106
retries=None,
107
inst_data: Optional[ConfigurableClassData] = None,
108
k8s_client_batch_api=None,
109
env_config_maps=None,
110
env_secrets=None,
111
volume_mounts=None,
112
volumes=None,
113
service_account_name=None,
114
image_pull_policy=None,
115
image_pull_secrets=None,
116
labels=None,
117
fail_pod_on_run_failure=None,
118
job_namespace=None,
119
):
120
"""
121
Initialize the CeleryK8sRunLauncher.
122
123
Args:
124
instance_config_map: Name of the ConfigMap containing instance configuration
125
dagster_home: Path to Dagster home directory
126
postgres_password_secret: Name of secret containing PostgreSQL password
127
load_incluster_config: Whether to load Kubernetes config from within cluster
128
kubeconfig_file: Path to kubeconfig file if not using in-cluster config
129
broker: Celery broker URL
130
backend: Celery backend URL
131
include: List of modules for Celery workers to import
132
config_source: Additional Celery configuration
133
retries: Retry configuration
134
inst_data: Configurable class instance data
135
k8s_client_batch_api: Override for Kubernetes batch API client
136
env_config_maps: List of ConfigMaps to mount as environment variables
137
env_secrets: List of Secrets to mount as environment variables
138
volume_mounts: List of volume mounts for pods
139
volumes: List of volumes for pods
140
service_account_name: Kubernetes service account name
141
image_pull_policy: Image pull policy for containers
142
image_pull_secrets: List of image pull secrets
143
labels: Labels to apply to Kubernetes resources
144
fail_pod_on_run_failure: Whether to fail pod on run failure
145
job_namespace: Kubernetes namespace for jobs
146
"""
147
148
def launch_run(self, context: LaunchRunContext) -> None:
149
"""
150
Launch a Dagster run as a Kubernetes job.
151
152
Args:
153
context: Launch context containing run information and code origin
154
"""
155
156
def terminate(self, run_id):
157
"""
158
Terminate a running Dagster job.
159
160
Args:
161
run_id: ID of the run to terminate
162
163
Returns:
164
True if termination was successful, False if run is already finished,
165
None if an exception occurred during termination
166
"""
167
168
def check_run_worker_health(self, run: DagsterRun) -> CheckRunHealthResult:
169
"""
170
Check the health of a run worker.
171
172
Args:
173
run: The Dagster run to check
174
175
Returns:
176
Health check result with worker status
177
"""
178
179
@classmethod
180
def config_type(cls):
181
"""Return the configuration schema for this run launcher."""
182
183
@classmethod
184
def from_config_value(cls, inst_data, config_value):
185
"""Create instance from configuration values."""
186
187
def get_k8s_job_config(self, job_image, exc_config) -> DagsterK8sJobConfig:
188
"""
189
Get Kubernetes job configuration.
190
191
Args:
192
job_image: Docker image for the job
193
exc_config: Executor configuration dictionary
194
195
Returns:
196
Kubernetes job configuration object
197
"""
198
199
def get_namespace_from_run_config(self, run_id: str) -> str:
200
"""
201
Extract namespace from run configuration.
202
203
Args:
204
run_id: Run identifier
205
206
Returns:
207
Kubernetes namespace name
208
"""
209
210
@property
211
def supports_check_run_worker_health(self) -> bool:
212
"""Whether this launcher supports health checks."""
213
214
@property
215
def inst_data(self) -> Optional[ConfigurableClassData]:
216
"""Configuration instance data."""
217
```
218
219
### Executor
220
221
The executor function that creates the distributed execution engine.
222
223
```python { .api }
224
@executor(
225
name="celery-k8s",
226
config_schema=celery_k8s_executor_config(),
227
requirements=multiple_process_executor_requirements(),
228
)
229
def celery_k8s_job_executor(init_context):
230
"""
231
Celery-based executor which launches tasks as Kubernetes Jobs.
232
233
The executor exposes config settings for the underlying Celery app and
234
Kubernetes job configuration. It works in concert with CeleryK8sRunLauncher
235
to provide distributed execution where:
236
237
1. A run worker Kubernetes Job traverses the dagster run execution plan
238
and submits steps to Celery queues for execution
239
2. Celery workers pick up step executions and spawn Kubernetes Jobs
240
for each step
241
242
Args:
243
init_context: Executor initialization context containing configuration
244
245
Returns:
246
CeleryK8sJobExecutor instance configured for distributed execution
247
"""
248
```
249
250
### Configuration
251
252
Configuration utilities for setting up the executor and launcher.
253
254
```python { .api }
255
def celery_k8s_executor_config():
256
"""
257
Return configuration schema for celery-k8s executor.
258
259
Merges Celery configuration, Kubernetes job configuration, and
260
additional celery-k8s specific options.
261
262
Returns:
263
Configuration dictionary with all available options
264
"""
265
266
def get_celery_engine_config(
267
image_pull_policy=None,
268
additional_env_config_maps=None
269
):
270
"""
271
Get engine configuration for Celery K8s execution.
272
273
Args:
274
image_pull_policy: Kubernetes image pull policy
275
additional_env_config_maps: Additional ConfigMaps for environment variables
276
277
Returns:
278
Engine configuration dictionary
279
"""
280
281
def get_celery_engine_job_config(
282
image_pull_policy=None,
283
additional_env_config_maps=None
284
):
285
"""
286
Get job configuration for Celery K8s execution.
287
288
Args:
289
image_pull_policy: Kubernetes image pull policy
290
additional_env_config_maps: Additional ConfigMaps for environment variables
291
292
Returns:
293
Job configuration dictionary
294
"""
295
```
296
297
### CeleryK8sJobExecutor
298
299
The executor implementation class that handles distributed step execution.
300
301
```python { .api }
302
class CeleryK8sJobExecutor(Executor):
303
"""
304
Executor that runs steps as Kubernetes Jobs via Celery task queuing.
305
306
This executor works in concert with CeleryK8sRunLauncher to provide
307
distributed execution of Dagster steps.
308
"""
309
310
def __init__(
311
self,
312
retries,
313
broker=None,
314
backend=None,
315
include=None,
316
config_source=None,
317
job_config=None,
318
job_namespace=None,
319
load_incluster_config=False,
320
kubeconfig_file=None,
321
repo_location_name=None,
322
job_wait_timeout=None,
323
per_step_k8s_config=None,
324
):
325
"""
326
Initialize the CeleryK8sJobExecutor.
327
328
Args:
329
retries: Retry mode configuration
330
broker: Celery broker URL
331
backend: Celery backend URL
332
include: List of modules for Celery workers to import
333
config_source: Additional Celery configuration dictionary
334
job_config: Kubernetes job configuration object
335
job_namespace: Kubernetes namespace for jobs
336
load_incluster_config: Whether to load Kubernetes config from within cluster
337
kubeconfig_file: Path to kubeconfig file
338
repo_location_name: Repository location name for execution
339
job_wait_timeout: Timeout for job completion in seconds
340
per_step_k8s_config: Per-step Kubernetes configuration overrides
341
"""
342
343
def execute(self, plan_context, execution_plan):
344
"""
345
Execute the execution plan using Celery task distribution.
346
347
Args:
348
plan_context: Pipeline execution context
349
execution_plan: Execution plan to run
350
351
Returns:
352
Generator of execution events
353
"""
354
355
def app_args(self):
356
"""
357
Return arguments for Celery app configuration.
358
359
Returns:
360
Dictionary with broker, backend, include, config_source, and retries
361
"""
362
363
@property
364
def retries(self):
365
"""Return retry mode configuration."""
366
```
367
368
### Celery Application
369
370
Pre-configured Celery application for K8s job execution.
371
372
```python { .api }
373
app: celery.Celery
374
"""Pre-configured Celery application with task routes for K8s job execution."""
375
376
execute_step_k8s_job: celery.Task
377
"""Celery task instance for executing steps as K8s jobs."""
378
```
379
380
### Constants
381
382
```python { .api }
383
CELERY_K8S_CONFIG_KEY = "celery-k8s"
384
"""Configuration key for celery-k8s executor."""
385
386
__version__ = "0.27.9"
387
"""Package version."""
388
```
389
390
## Configuration Options
391
392
The executor supports extensive configuration for both Celery and Kubernetes:
393
394
### Kubernetes Configuration
395
- `job_image`: Docker image for step execution jobs
396
- `job_namespace`: Kubernetes namespace for jobs (default: "default")
397
- `load_incluster_config`: Load K8s config from within cluster (default: True)
398
- `kubeconfig_file`: Path to kubeconfig file
399
- `job_wait_timeout`: Timeout for job completion (default: 4 hours)
400
- `env_config_maps`: ConfigMaps to mount as environment variables
401
- `env_secrets`: Secrets to mount as environment variables
402
- `volume_mounts`: Volume mounts for job pods
403
- `volumes`: Volumes for job pods
404
- `service_account_name`: Kubernetes service account
405
- `image_pull_policy`: Image pull policy (default: "IfNotPresent")
406
- `image_pull_secrets`: Image pull secrets
407
- `labels`: Labels for Kubernetes resources
408
409
### Celery Configuration
410
- `broker`: Celery broker URL (e.g., "pyamqp://guest@localhost//")
411
- `backend`: Celery results backend URL (e.g., "rpc://")
412
- `include`: Modules for Celery workers to import
413
- `config_source`: Additional Celery configuration dictionary
414
- `retries`: Retry configuration for failed tasks
415
416
### Per-Step Configuration
417
- `per_step_k8s_config`: Per-operation Kubernetes configuration overrides
418
419
## Usage Examples
420
421
### Basic Job Definition
422
423
```python
424
from dagster import op, job
425
from dagster_celery_k8s import celery_k8s_job_executor
426
427
@op
428
def process_data():
429
return "processed"
430
431
@op
432
def save_results(data: str):
433
print(f"Saving: {data}")
434
435
@job(executor_def=celery_k8s_job_executor)
436
def my_pipeline():
437
save_results(process_data())
438
```
439
440
### Advanced Configuration
441
442
```python
443
from dagster import job, RunConfig
444
from dagster_celery_k8s import celery_k8s_job_executor
445
446
@job(executor_def=celery_k8s_job_executor)
447
def advanced_pipeline():
448
# Job operations
449
pass
450
451
# Run with specific configuration
452
run_config = RunConfig(
453
execution={
454
"config": {
455
"job_image": "my-registry/my-image:v1.0.0",
456
"job_namespace": "data-processing",
457
"broker": "redis://redis-service:6379/0",
458
"backend": "redis://redis-service:6379/1",
459
"job_wait_timeout": 3600, # 1 hour
460
"per_step_k8s_config": {
461
"heavy_computation": {
462
"container_config": {
463
"resources": {
464
"requests": {"cpu": "2", "memory": "4Gi"},
465
"limits": {"cpu": "4", "memory": "8Gi"}
466
}
467
}
468
}
469
}
470
}
471
}
472
)
473
```
474
475
### Instance Configuration
476
477
Configure in your `dagster.yaml`:
478
479
```yaml
480
run_launcher:
481
module: dagster_k8s.launcher
482
class: CeleryK8sRunLauncher
483
config:
484
# Required configuration
485
instance_config_map: "dagster-instance-config"
486
dagster_home: "/opt/dagster/dagster_home"
487
postgres_password_secret: "dagster-postgres-secret"
488
489
# Celery configuration
490
broker: "redis://redis-service:6379/0"
491
backend: "redis://redis-service:6379/1"
492
493
# Kubernetes configuration
494
job_namespace: "dagster-execution"
495
service_account_name: "dagster-service-account"
496
image_pull_policy: "Always"
497
498
# Resource configuration
499
env_config_maps:
500
- "dagster-env-config"
501
env_secrets:
502
- "dagster-secrets"
503
504
labels:
505
team: "data-engineering"
506
environment: "production"
507
```
508
509
## Error Handling
510
511
The package provides comprehensive error handling for:
512
513
### Kubernetes Errors
514
- Job creation failures and conflicts
515
- Pod scheduling issues
516
- Resource allocation problems
517
- Network connectivity issues
518
- Timeout scenarios
519
520
### Celery Errors
521
- Broker connection failures
522
- Task serialization issues
523
- Worker availability problems
524
- Message queue overflow
525
526
### Common Error Patterns
527
528
```python
529
from dagster import DagsterInvariantViolationError
530
531
# Image configuration error
532
if not job_image:
533
raise DagsterInvariantViolationError(
534
"You have not specified a job_image in your executor configuration. "
535
"Specify the job_image in the executor config section."
536
)
537
538
# Compatibility error
539
if not isinstance(run_launcher, CeleryK8sRunLauncher):
540
raise DagsterUnmetExecutorRequirementsError(
541
"This executor is only compatible with a CeleryK8sRunLauncher; "
542
"configure the CeleryK8sRunLauncher on your instance to use it."
543
)
544
```
545
546
## Integration Requirements
547
548
### Dependencies
549
- dagster==1.11.9
550
- dagster-k8s==0.27.9
551
- dagster-celery==0.27.9
552
- kubernetes (Python client)
553
554
### Environment Setup
555
1. **Kubernetes Cluster**: Access to a Kubernetes cluster
556
2. **Celery Broker**: Redis or RabbitMQ broker service
557
3. **Dagster Instance**: PostgreSQL-backed Dagster instance
558
4. **Container Registry**: Access to push/pull Docker images
559
5. **RBAC**: Kubernetes permissions for job creation and management
560
561
### Deployment Considerations
562
- Celery workers must be deployed with `-A dagster_celery_k8s.app` argument
563
- Kubernetes Jobs require appropriate resource quotas and limits
564
- Network policies must allow communication between components
565
- Persistent volumes may be needed for data sharing between steps