1---
2title: Ingest events into BigQuery
3description: A Cloud Run service publishes to Pub/Sub, a Cloud Run Job drains the subscription into BigQuery, and one host triggers the other.
4---
5
6Analytics ingestion has a shape that predates the cloud: accept the
7event fast, queue it, and write it to the warehouse in batches.
8Nothing on the request path touches BigQuery, so a schema change or a
9slow warehouse cannot take the ingest endpoint down — the messages
10just queue up.
11
12On GCP that is three resources and two hosts:
13
14| Piece | What it is | Why |
15| --- | --- | --- |
16| `PubSub.Topic` + pull `Subscription` | the queue | holds events until they are acked |
17| `GCP.Function` (Cloud Run Service) | the front door | accepts and publishes, scales to zero |
18| `GCP.Run.Job` | the drain | runs to completion, nothing billed between batches |
19| `BigQuery.Dataset` + `Table` | the warehouse | what analysts query |
20
21This guide walks through `examples/gcp-event-pipeline`.
22
23## The queue and the warehouse
24
25Resources that depend on another resource are built inside an
26`Effect.gen`, because the topic has to exist before its name can be
27referenced:
28
29```typescript
30// src/resources.ts
31export const Events = GCP.PubSub.Topic("Events", {});
32
33export const Analytics = GCP.BigQuery.Dataset("Analytics", {
34 location: "US-CENTRAL1",
35 forceDestroy: true,
36});
37
38export const Inbox = Effect.gen(function* () {
39 const topic = yield* Events;
40 return yield* GCP.PubSub.Subscription("Inbox", {
41 topic: topic.name,
42 ackDeadlineSeconds: 60,
43 });
44});
45
46export const EventsTable = Effect.gen(function* () {
47 const dataset = yield* Analytics;
48 return yield* GCP.BigQuery.Table("EventsTable", {
49 datasetId: dataset.datasetId,
50 tableId: "events",
51 schema: [
52 { name: "id", type: "STRING", mode: "REQUIRED" },
53 { name: "type", type: "STRING", mode: "REQUIRED" },
54 { name: "occurredAt", type: "TIMESTAMP", mode: "REQUIRED" },
55 { name: "payload", type: "JSON" },
56 ],
57 });
58});
59```
60
61Yielding `Inbox` from two different hosts is free: resources are keyed
62by logical id, so both get the same subscription rather than two.
63
64`payload` is a single JSON column instead of a column per attribute,
65so a producer can add a field without a schema migration.
66
67## The front door
68
69```typescript
70// src/Ingest.ts
71export default class Ingest extends GCP.Function<Ingest>()(
72 "Ingest",
73 {
74 main: import.meta.url,
75 location: "us-central1",
76 invokerIamDisabled: true,
77 },
78 Effect.gen(function* () {
79 const topic = yield* Events;
80 const table = yield* EventsTable;
81 const drain = yield* Drain;
82
83 // pubsub.publisher on the topic only.
84 const publisher = yield* GCP.PubSub.WriteTopic(topic);
85 // bigquery.dataViewer on the table, plus bigquery.jobUser on the
86 // project — BigQuery only grants running query jobs there.
87 const warehouse = yield* GCP.BigQuery.ReadTable(table);
88 // Binding a Job to a Service grants run.jobsExecutorWithOverrides on
89 // the job — this is how one host triggers another.
90 const runDrain = yield* GCP.Run.RunJob(drain);
91
92 // An accessor: the table id is bound at deploy time and read back
93 // inside the handler.
94 const tableId = yield* table.tableId;
95```
96
97The two data bindings are access-level clients: the service may
98publish to the topic and read the table, and nothing else. It cannot
99pull from the subscription or insert rows — that is the drain's job.
100
101`GCP.Run.RunJob(drain)` is the interesting one: binding a Job to a
102Service grants `roles/run.jobsExecutorWithOverrides` on that one Job
103to the service's runtime account and returns a callable. That is how
104one host triggers another, with no job name to interpolate and no IAM
105policy to write.
106
107### `POST /events`
108
109```typescript
110 const event: EventRow = {
111 id: crypto.randomUUID(),
112 type: body.type,
113 occurredAt: new Date().toISOString(),
114 payload: JSON.stringify(body.payload ?? {}),
115 };
116
117 yield* publisher
118 .publish({
119 data: JSON.stringify(event),
120 // Attributes are queryable without decoding the body,
121 // which lets a filtered subscription fan out by type.
122 attributes: { type: event.type },
123 })
124 .pipe(Effect.orDie);
125
126 return yield* HttpServerResponse.json(
127 { id: event.id },
128 { status: 202 },
129 );
130```
131
132`publish` takes a string or bytes and handles Pub/Sub's base64
133encoding; it returns the server-assigned message id.
134
135`202` rather than `201`: the event is accepted, not yet stored.
136
137### `POST /drain`
138
139```typescript
140 if (request.method === "POST" && url.pathname === "/drain") {
141 const operation = yield* runDrain().pipe(Effect.orDie);
142 return yield* HttpServerResponse.json(
143 { execution: operation.name ?? null },
144 { status: 202 },
145 );
146 }
147```
148
149Starting a Cloud Run Job returns as soon as the execution is created,
150not when it finishes. In production, Cloud Scheduler would call this
151on a cron; the route exists so you can force a batch by hand.
152
153### `GET /events/count`
154
155```typescript
156 const events = yield* tableId;
157 // Unqualified table names resolve against the bound table's
158 // dataset; `params` become named `@type` parameters.
159 const rows = yield* warehouse
160 .query(
161 type
162 ? `SELECT COUNT(*) AS n FROM \`${events}\` WHERE type = @type`
163 : `SELECT COUNT(*) AS n FROM \`${events}\``,
164 type ? { type } : undefined,
165 )
166 .pipe(Effect.orDie);
167
168 return yield* HttpServerResponse.json({
169 count: Number(rows[0]?.n ?? 0),
170 });
171```
172
173`ReadTable.query` sets `defaultDataset` from the bound table, so the
174bare table name resolves, and returns rows decoded to plain JavaScript
175(`INT64` → `number`). Named parameters keep a caller's `?type=` out of
176the SQL text.
177
178## The drain
179
180`GCP.Run.Job` is the same platform with a `run` entry instead of
181`fetch`. It starts, does the work, and exits.
182
183```typescript
184// src/Drain.ts
185export default class Drain extends GCP.Run.Job<Drain>()(
186 "Drain",
187 {
188 main: import.meta.url,
189 location: "us-central1",
190 },
191 Effect.gen(function* () {
192 const inbox = yield* Inbox;
193 const table = yield* EventsTable;
194
195 // pubsub.subscriber on the subscription, and bigquery.dataEditor on
196 // the table — nothing on the project.
197 const subscription = yield* GCP.PubSub.ReadSubscription(inbox);
198 const warehouse = yield* GCP.BigQuery.WriteTable(table);
199
200 return {
201 // A pull may return fewer messages than are waiting, so drain
202 // batch by batch until one comes back empty.
203 run: Effect.gen(function* () {
204 // A pull waits for messages; an empty subscription answers
205 // nothing, so a quiet 10 seconds means the backlog is drained.
206 const received = yield* subscription
207 .pull({ maxMessages: BATCH })
208 .pipe(Effect.timeoutOption("10 seconds"));
209 const messages = Option.getOrElse(received, () => []);
210 if (messages.length === 0) {
211 yield* Effect.log("drain: nothing left");
212 return 0;
213 }
214
215 const rows = messages.map(
216 (message) => JSON.parse(message.text) as EventRow,
217 );
218
219 // insertIds make the streaming insert idempotent inside
220 // BigQuery's dedup window, so a redelivered batch collapses.
221 yield* warehouse.insert(rows, { insertIds: rows.map((row) => row.id) });
222
223 yield* subscription.acknowledge(
224 messages.map((message) => message.ackId),
225 );
226
227 yield* Effect.log(`drain: wrote ${rows.length} row(s)`);
228 return messages.length;
229 }).pipe(
230 Effect.repeat({ until: (count) => count === 0 }),
231 Effect.asVoid,
232 Effect.orDie,
233 ),
234 };
235 }).pipe(
236 Effect.provide([
237 GCP.PubSub.ReadSubscriptionHttp,
238 GCP.BigQuery.WriteTableHttp,
239 ]),
240 ),
241) {}
242```
243
244`pull` decodes each message for you (`text`, `data`, `attributes`,
245`ackId`), and `insert` takes plain row objects. Because the table
246depends on the dataset, binding the table is enough to make the job
247wait for both before its first run.
248
249:::note[Insert first, ack second]
250The order is deliberate. A crash between the two redelivers the batch
251and duplicates rows, which `SELECT DISTINCT id` removes. Acking first
252would lose the events outright. `insertIds` make the streaming insert
253idempotent inside BigQuery's dedup window, so most redeliveries
254collapse on their own.
255:::
256
257## Try it
258
259Both hosts are built from `main`, so Docker has to be running.
260
261```sh
262cd examples/gcp-event-pipeline
263bun alchemy deploy
264```
265
266```sh
267curl -X POST "$URL/events" -H 'content-type: application/json' \
268 -d '{"type":"signup","payload":{"plan":"pro"}}'
269# → { "id": "…" } (202)
270
271curl "$URL/events/count?type=signup" # → { "count": 0 }
272curl -X POST "$URL/drain" # → { "execution": "…" } (202)
273# a few seconds later
274curl "$URL/events/count?type=signup" # → { "count": 1 }
275```
276
277`test/integ.test.ts` publishes three events under a unique type,
278asserts BigQuery has not seen them, starts the drain, and polls the
279count until the rows land.
280
281## Where to take it
282
283- **Schedule the drain.** `GCP.CloudScheduler.Job` calling the same
284 `run.jobs.run` endpoint turns this into a batch pipeline with no
285 manual step.
286- **Fan out by type.** A second `Subscription` with a filter on the
287 `type` attribute routes a subset of events elsewhere without
288 touching the producer.
289- **Push instead of pull.** Set `pushConfig.pushEndpoint` on the
290 subscription to a Cloud Run URL if you want per-message delivery
291 rather than batches.
292
293## Next
294
295- [Serve an API on Cloud Run](/gcp/guides/cloud-run-api) — the
296 request-path counterpart.
297- [How bindings grant IAM](/gcp/guides/bindings).
298- [PubSub.Topic](/providers/gcp/pubsub/topic),
299 [PubSub.Subscription](/providers/gcp/pubsub/subscription),
300 [BigQuery.Table](/providers/gcp/bigquery/table),
301 [Run.Job](/providers/gcp/run/job) reference.