0
# Dagster GCP
1
2
Google Cloud Platform integration components for the Dagster data orchestration framework. This package provides resources, I/O managers, operations, and utilities for building data pipelines that leverage GCP services including BigQuery, Google Cloud Storage, Dataproc, and Pipes for external process communication.
3
4
## Package Information
5
6
- **Package Name**: dagster-gcp
7
- **Language**: Python
8
- **Installation**: `pip install dagster-gcp`
9
10
## Core Imports
11
12
```python
13
import dagster_gcp
14
```
15
16
Common specific imports:
17
18
```python
19
from dagster_gcp import (
20
BigQueryResource, BigQueryIOManager,
21
GCSResource, GCSPickleIOManager,
22
DataprocResource, configurable_dataproc_op
23
)
24
```
25
26
Pipes components (imported from submodule):
27
28
```python
29
from dagster_gcp.pipes import (
30
PipesDataprocJobClient,
31
PipesGCSContextInjector,
32
PipesGCSMessageReader,
33
PipesGCSLogReader
34
)
35
```
36
37
## Basic Usage
38
39
```python
40
from dagster import asset, Definitions
41
from dagster_gcp import BigQueryResource, GCSResource, GCSPickleIOManager
42
43
@asset
44
def my_data_asset(bigquery: BigQueryResource):
45
with bigquery.get_client() as client:
46
query = "SELECT * FROM `project.dataset.table` LIMIT 100"
47
df = client.query(query).to_dataframe()
48
return df
49
50
defs = Definitions(
51
assets=[my_data_asset],
52
resources={
53
"bigquery": BigQueryResource(project="my-gcp-project"),
54
"gcs": GCSResource(project="my-gcp-project"),
55
"io_manager": GCSPickleIOManager(
56
gcs_bucket="my-data-bucket",
57
gcs=GCSResource(project="my-gcp-project")
58
)
59
}
60
)
61
```
62
63
## Architecture
64
65
The dagster-gcp package follows Dagster's resource and I/O manager patterns while providing GCP-specific implementations:
66
67
- **Resources**: Configurable connections to GCP services (BigQuery, GCS, Dataproc)
68
- **I/O Managers**: Data storage and retrieval using GCP storage services
69
- **Operations**: Pre-built ops for common GCP tasks (data loading, job execution)
70
- **Pipes Integration**: External process communication through GCP services
71
72
Each integration supports both modern ConfigurableResource patterns and legacy resource factory functions for backward compatibility.
73
74
## Capabilities
75
76
### BigQuery Integration
77
78
Data warehousing operations including I/O managers for BigQuery tables, operations for data loading and querying, and resources for BigQuery client management with authentication support.
79
80
```python { .api }
81
class BigQueryResource(ConfigurableResource):
82
project: Optional[str]
83
location: Optional[str]
84
gcp_credentials: Optional[str]
85
86
def get_client(self) -> Iterator[bigquery.Client]: ...
87
88
class BigQueryIOManager(ConfigurableIOManagerFactory):
89
project: str
90
dataset: Optional[str]
91
location: Optional[str]
92
gcp_credentials: Optional[str]
93
temporary_gcs_bucket: Optional[str]
94
timeout: Optional[float]
95
```
96
97
[BigQuery Integration](./bigquery.md)
98
99
### Google Cloud Storage (GCS)
100
101
File storage and management including I/O managers for pickled objects, file managers for direct GCS operations, compute log management, and sensor utilities for GCS-based data processing.
102
103
```python { .api }
104
class GCSResource(ConfigurableResource):
105
project: Optional[str]
106
107
def get_client(self) -> storage.Client: ...
108
109
class GCSPickleIOManager(ConfigurableIOManager):
110
gcs: ResourceDependency[GCSResource]
111
gcs_bucket: str
112
gcs_prefix: str = "dagster"
113
114
def load_input(self, context) -> Any: ...
115
def handle_output(self, context, obj) -> None: ...
116
```
117
118
[Google Cloud Storage](./gcs.md)
119
120
### Dataproc Integration
121
122
Apache Spark cluster management and job execution including resources for Dataproc cluster lifecycle management, operations for submitting and monitoring Spark jobs, and comprehensive configuration support for cluster and job parameters.
123
124
```python { .api }
125
class DataprocResource(ConfigurableResource):
126
project_id: str
127
region: str
128
cluster_name: str
129
labels: Optional[dict[str, str]]
130
cluster_config_yaml_path: Optional[str]
131
cluster_config_json_path: Optional[str]
132
cluster_config_dict: Optional[dict]
133
134
def get_client(self) -> DataprocClient: ...
135
136
@op
137
def configurable_dataproc_op(
138
dataproc: DataprocResource,
139
config: DataprocOpConfig
140
) -> Any: ...
141
```
142
143
[Dataproc Integration](./dataproc.md)
144
145
### Pipes Integration
146
147
External process communication through GCP services including clients for running workloads on Dataproc, context injectors for passing data via GCS, and message readers for collecting results and logs from external processes.
148
149
```python { .api }
150
class PipesDataprocJobClient(PipesClient):
151
client: JobControllerClient
152
context_injector: PipesContextInjector
153
message_reader: PipesMessageReader
154
forward_termination: bool = True
155
poll_interval: float = 5.0
156
157
def run(self, context, submit_job_params, extras) -> PipesClientCompletedInvocation: ...
158
159
class PipesGCSContextInjector(PipesContextInjector):
160
bucket: str
161
client: GCSClient
162
key_prefix: Optional[str]
163
```
164
165
[Pipes Integration](./pipes.md)
166
167
## Types
168
169
### Core Types
170
171
```python { .api }
172
# File handle for GCS objects
173
class GCSFileHandle(FileHandle):
174
@property
175
def gcs_bucket(self) -> str: ...
176
177
@property
178
def gcs_key(self) -> str: ...
179
180
@property
181
def gcs_path(self) -> str: ...
182
183
@property
184
def path_desc(self) -> str: ...
185
186
# BigQuery error handling
187
class BigQueryError(Exception): ...
188
189
# Dataproc error handling
190
class DataprocError(Exception): ...
191
```