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
# Parent/child DAG: fetch the users list once, then for each user fan out
# to fetch that user's posts and orders in parallel. `${users.id}` is
# resolved per parent record at runtime, so each child invocation gets
# its own URL and its own output file.
#
# Required env vars:
# API_TOKEN — bearer token for the source API
version: 1
name: rest_users_posts_dag
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: 100
page_size_param: per_page
max_pages: 50
timeout: 30
max_retries: 3
retry_backoff: 1
tolerated_http_errors: []
replication_method:
type: FullTable
partitions: []
schema_sample_size: 100
sink:
type: jsonl
config:
append: false
pretty: false
matrix:
# Root row: fetch the users list.
- id: users
source:
config:
path: /v1/users
name: users
records_path: $.users[*]
primary_keys: [id]
sink:
config:
path: ./out/users.jsonl
# Child row: one fetch per user, populated from `${users.id}`.
- id: user_posts
parent: users
parent_key: id
source:
config:
path: /v1/users/${users.id}/posts
name: user_posts
records_path: $.posts[*]
primary_keys: [id]
sink:
config:
path: ./out/posts-${users.id}.jsonl
# Second child row: also one fetch per user, but for orders.
- id: user_orders
parent: users
parent_key: id
source:
config:
path: /v1/users/${users.id}/orders
name: user_orders
records_path: $.orders[*]
primary_keys: [order_id]
sink:
config:
path: ./out/orders-${users.id}.jsonl
execution:
max_concurrent: 8
on_error: continue