1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
# Three independent REST endpoints → three BigQuery tables, sharing one
# base source config and one base sink config. Each row varies only in
# the REST `path` and the BigQuery `(dataset_id, table_id)`. No DAG —
# all three rows are root rows and run concurrently (bounded by
# `execution.max_concurrent`).
#
# Required env vars:
# API_TOKEN — bearer token for the source API
# GCP_KEY_JSON — inline service-account JSON
version: 1
name: rest_to_bigquery_matrix
pipeline:
source:
type: rest
config:
base_url: https://api.example.com
method: GET
auth:
type: bearer
config:
token: ${env:API_TOKEN}
pagination:
type: PageNumber
param_name: page
start_page: 1
page_size: 500
page_size_param: per_page
max_pages: 200
timeout: 30
max_retries: 5
retry_backoff: 2
tolerated_http_errors: []
replication_method:
type: FullTable
partitions: []
schema_sample_size: 100
sink:
type: bigquery
config:
project_id: my-gcp-project
auth:
type: service_account_key
config:
json: ${env:GCP_KEY_JSON}
batch_size: 1000
matrix:
- id: users
source:
config:
path: /v1/users
name: users
records_path: $.users[*]
primary_keys: [id]
sink:
config:
dataset_id: analytics
table_id: users
- id: products
source:
config:
path: /v1/products
name: products
records_path: $.products[*]
primary_keys: [sku]
sink:
config:
dataset_id: analytics
table_id: products
- id: orders
source:
config:
path: /v1/orders
name: orders
records_path: $.orders[*]
primary_keys: [order_id]
sink:
config:
dataset_id: analytics
table_id: orders
execution:
max_concurrent: 3
on_error: continue