Apache Airflow 數(shù)據(jù)編排實戰(zhàn)

第Ⅰ部分  入門
第1章  遇見Apache Airflow 3
1.1  數(shù)據(jù)管道介紹 3
1.1.1  數(shù)據(jù)管道的圖形表示 4
1.1.2  運行管道圖 5
1.1.3  管道圖與順序腳本 6
1.1.4  使用工作流管理器運行數(shù)據(jù)流 8
1.2  Airflow介紹 9
1.2.1  通過Python代碼靈活定義數(shù)據(jù)管道 9
1.2.2  調(diào)度并執(zhí)行數(shù)據(jù)管道 10
1.2.3  監(jiān)控和處理故障 11
1.2.4  增量載入和回填 14
1.3  何時使用Airflow 14
1.3.1  選擇Airflow的原因 14
1.3.2  不使用Airflow的理由 15
1.4  本書的其余部分 15
1.5  本章小結(jié) 16
第2章  Airflow DAG深度解析 17
2.1  從大量數(shù)據(jù)源中收集數(shù)據(jù) 17
2.2  編寫你的第一個Airflow DAG 19
2.2.1  任務(wù)與operator 22
2.2.2  運行任意Python代碼 23
2.3  在Airflow中運行DAG 25
2.3.1  在Python環(huán)境中運行Airflow 25
2.3.2  在Docker容器中運行Airflow 26
2.3.3  使用Airflow圖形界面 27
2.4  運行定時任務(wù) 31
2.5  處理失敗的任務(wù) 32
2.6  本章小結(jié) 34
第3章  Airflow中的調(diào)度 35
3.1  示例：處理用戶事件 35
3.2  定期執(zhí)行DAG 37
3.2.1  使用調(diào)度器計劃性運行 37
3.2.2  基于cron的時間間隔 38
3.2.3  基于頻率的時間間隔 40
3.3  增量處理數(shù)據(jù) 40
3.3.1  獲取增量事件數(shù)據(jù) 40
3.3.2  使用執(zhí)行日期的動態(tài)時間參考 42
3.3.3  對數(shù)據(jù)執(zhí)行分區(qū) 43
3.4  理解Airflow的執(zhí)行日期 45
3.5  使用回填技術(shù)填補過去的空白 47
3.6  任務(wù)設(shè)計的最佳實踐 49
3.6.1  原子性 49
3.6.2  冪等性 51
3.7  本章小結(jié) 52
第4章  使用Airflow context對任務(wù)進行模板化 53
4.1  為Airflow準(zhǔn)備數(shù)據(jù) 53
4.2  任務(wù)context和Jinja模板 55
4.2.1  對operator使用參數(shù)模板 56
4.2.2  模板中可用的變量及表達(dá)式 57
4.2.3  對PythonOperator使用模板 60
4.2.4  為PythonOperator提供變量 64
4.2.5  檢查模板化參數(shù) 66
4.3  連接到其他系統(tǒng) 67
4.4  本章小結(jié) 74
第5章  定義任務(wù)之間的依賴關(guān)系 75
5.1  基本依賴關(guān)系 75
5.1.1  線性依賴關(guān)系 75
5.1.2  扇入/扇出依賴 77
5.2  分支 79
5.2.1  在任務(wù)內(nèi)部執(zhí)行分支操作 79
5.2.2  在DAG中使用分支技術(shù) 81
5.3  帶有條件的任務(wù) 85
5.3.1  在任務(wù)內(nèi)部使用條件 85
5.3.2  對DAG使用條件 86
5.3.3  使用內(nèi)置operator 88
5.4  觸發(fā)條件詳解 88
5.4.1  什么是觸發(fā)規(guī)則 88
5.4.2  失敗的影響 89
5.4.3  其他觸發(fā)規(guī)則 90
5.5  在任務(wù)之間共享數(shù)據(jù) 91
5.5.1  使用XCom共享數(shù)據(jù) 91
5.5.2  XCom的適用場景 94
5.5.3  使用自定義XCom后端存儲 95
5.6  使用Taskflow API連接Python任務(wù) 95
5.6.1  使用Taskflow API簡化Python任務(wù) 96
5.6.2  Taskflow API的適用場景 98
5.7  本章小結(jié) 99
第Ⅱ部分  Airflow深入學(xué)習(xí)
第6章  觸發(fā)工作流 103
6.1  帶有傳感器的輪詢條件 103
6.1.1  輪詢自定義條件 106
6.1.2  傳感器的異常情況 107
6.2  觸發(fā)其他DAG 110
6.2.1  使用TriggerDagRunOperator執(zhí)行回填操作 114
6.2.2  輪詢其他 DAG 的狀態(tài) 114
6.3  使用REST/CLI啟動工作流 117
6.4  本章小結(jié) 120
第7章  與外部系統(tǒng)通信 121
7.1  連接到云服務(wù) 122
7.1.1  安裝額外的依賴軟件包 122
7.1.2  開發(fā)一個機器學(xué)習(xí)模型 123
7.1.3  在本地開發(fā)外部系統(tǒng)程序 128
7.2  在系統(tǒng)之間移動數(shù)據(jù) 134
7.2.1  實現(xiàn)PostgresToS3Operator 136
7.2.2  將繁重的任務(wù)“外包”出去 139
7.3  本章小結(jié) 141
第8章  創(chuàng)建自定義組件 143
8.1  從PythonOperator開始 143
8.1.1  模擬電影評分API 144
8.1.2  從API獲取評分?jǐn)?shù)據(jù) 146
8.1.3  構(gòu)建具體的DAG 149
8.2  創(chuàng)建自定義hook 151
8.2.1  設(shè)定自定義hook 151
8.2.2  使用MovielensHook構(gòu)建DAG 156
8.3  構(gòu)建自定義operator 158
8.3.1  創(chuàng)建自定義operator 158
8.3.2  創(chuàng)建用于獲取評分?jǐn)?shù)據(jù)的operator 159
8.4  創(chuàng)建自定義傳感器 162
8.5  將你的組件打包 165
8.5.1  引導(dǎo)Python包 166
8.5.2  安裝你的Python包 168
8.6  本章小結(jié) 169
第9章  測試 171
9.1  開始測試 171
9.1.1  所有DAG的完整性測試 172
9.1.2  設(shè)置CI/CD管道 177
9.1.3  編寫單元測試 179
9.1.4  pytest項目結(jié)構(gòu) 180
9.1.5  使用磁盤上的文件測試 184
9.2  在測試中使用DAG和任務(wù)context 186
9.3  使用測試進行開發(fā) 198
9.4  使用Whirl模擬生產(chǎn)環(huán)境 201
9.5  創(chuàng)建DTAP環(huán)境 201
9.6  本章小結(jié) 201
第10章  在容器中運行任務(wù) 203
10.1  同時使用多個不同operator
所面臨的挑戰(zhàn) 203
10.1.1  operator接口和實現(xiàn) 204
10.1.2  復(fù)雜且相互沖突的依賴關(guān)系 204
10.1.3  轉(zhuǎn)向通用operator 205
10.2  容器 205
10.2.1  什么是容器 206
10.2.2  運行第一個Docker容器 207
10.2.3  創(chuàng)建Docker映像 207
10.2.4  使用卷持久化數(shù)據(jù) 209
10.3  容器與Airflow 212
10.3.1  容器中的任務(wù) 212
10.3.2  為什么使用容器 212
10.4  在Docker中運行任務(wù) 213
10.4.1 使用DockerOperator 213
10.4.2  為任務(wù)創(chuàng)建容器映像 215
10.4.3  使用Docker任務(wù)創(chuàng)建DAG 218
10.4.4  基于Docker的工作流 220
10.5  在Kubernetes中運行任務(wù) 221
10.5.1  Kubernetes介紹 221
10.5.2  設(shè)置Kubernetes 222
10.5.3  使用KubernetesPodOperator 225
10.5.4  診斷Kubernetes相關(guān)的問題 228
10.5.5  與基于docker的工作流的區(qū)別 230
10.6  本章小結(jié) 231
第Ⅲ部分  Airflow實踐
第11章  最佳實現(xiàn) 235
11.1  編寫清晰的DAG 235
11.1.1  使用風(fēng)格約定 235
11.1.2  集中管理憑證 239
11.1.3  統(tǒng)一指定配置詳細(xì)信息 240
11.1.4  避免在DAG定義中計算 242
11.1.5  使用工廠函數(shù)生成通用模式 244
11.1.6  使用任務(wù)組對相關(guān)任務(wù)進行分組 247
11.1.7  為重大變更創(chuàng)建新的DAG 248
11.2  設(shè)計可重用的任務(wù) 249
11.2.1  要求任務(wù)始終滿足冪等性 249
11.2.2  任務(wù)結(jié)果的確定性 249
11.2.3  使用函數(shù)式范式設(shè)計任務(wù) 250
11.3  高效處理數(shù)據(jù) 250
11.3.1  限制處理的數(shù)據(jù)量 250
11.3.2  增量載入與增量處理 252
11.3.3  緩存中間數(shù)據(jù) 252
11.3.4  不要將數(shù)據(jù)存儲在本地文件系統(tǒng) 253
11.3.5  將工作卸載到外部系統(tǒng)或源系統(tǒng) 253
11.4  管理資源 254
11.4.1  使用資源池管理并發(fā) 254
11.4.2  使用SLA和告警來檢測長時間運行的任務(wù) 255
11.5  本章小結(jié) 256
第12章  在生產(chǎn)環(huán)境中使用Airflow 257
12.1  Airflow架構(gòu) 258
12.1.1  挑選適合的執(zhí)行器 259
12.1.2  為Airflow配置metastore 259
12.1.3  深入了解調(diào)度器 261
12.2  安裝每個執(zhí)行器 265
12.2.1  設(shè)置SequentialExecutor 266
12.2.2  設(shè)置LocalExecutor 266
12.2.3  設(shè)置CeleryExecutor 267
12.2.4  設(shè)置KubernetesExecutor 269
12.3  捕獲所有Airflow進程的日志 276
12.3.1  捕獲Web服務(wù)器輸出 276
12.3.2  捕獲調(diào)度器輸出 277
12.3.3  捕獲任務(wù)日志 278
12.3.4  將日志發(fā)送到遠(yuǎn)程存儲 278
12.4  可視化及監(jiān)控Airflow指標(biāo) 279
12.4.1  從Airflow收集指標(biāo) 279
12.4.2  配置Airflow以發(fā)送指標(biāo) 280
12.4.3  配置Prometheus以收集指標(biāo) 281
12.4.4  使用Grafana創(chuàng)建儀表板 283
12.4.5  應(yīng)監(jiān)控的指標(biāo) 285
12.5  如何獲得失敗任務(wù)的通知 287
12.5.1  DAG和operator內(nèi)的告警 287
12.5.2  定義服務(wù)級別協(xié)議(SLA) 289
12.6  可伸縮性與性能 290
12.6.1  控制最大運行任務(wù)數(shù) 290
12.6.2  系統(tǒng)性能配置 292
12.6.3  運行多個調(diào)度器 292
12.7  本章小結(jié) 293
第13章  Airflow安全性 295
13.1  保護Airflow Web界面 296
13.1.1  將用戶添加到RBAC界面 296
13.1.2  配置RBAC界面 299
13.2  加密靜態(tài)數(shù)據(jù) 300
13.3  連接LDAP服務(wù) 301
13.3.1  理解LDAP 302
13.3.2  從LDAP服務(wù)獲取用戶 304
13.4  加密與Web服務(wù)器的通信 305
13.4.1  了解HTTPS 305
13.4.2  為HTTPS配置證書 307
13.5  從認(rèn)證管理系統(tǒng)獲取憑證 311
13.6  本章小結(jié) 314
第14章  實戰(zhàn)：探索游覽紐約市的最快方式 315
14.1  理解數(shù)據(jù) 318
14.1.1  Yellow Cab文件共享 318
14.1.2  Citi Bike REST API 319
14.1.3  確定算法 320
14.2  提取數(shù)據(jù) 320
14.2.1  下載Citi Bike數(shù)據(jù) 321
14.2.2  下載Yellow Cab數(shù)據(jù) 323
14.3  對數(shù)據(jù)應(yīng)用類似的轉(zhuǎn)換 325
14.4  構(gòu)建數(shù)據(jù)管道 330
14.5  開發(fā)冪等的數(shù)據(jù)管道 331
14.6  本章小結(jié) 333
第Ⅳ部分  在云端
第15章  Airflow在云端 337
15.1  設(shè)計云端部署策略 337
15.2  云端專用的hook和operator 339
15.3  托管服務(wù) 340
15.3.1  Astronomer.io 340
15.3.2  Google Cloud Composer 340
15.3.3  適用于Apache Airflow的
Amazon托管工作流 341
15.4  選擇部署策略 342
15.5  本章小結(jié) 342
第16章  在AWS中運行Airflow 345
16.1  在AWS中部署Airflow 345
16.1.1  選擇云服務(wù) 345
16.1.2  設(shè)計網(wǎng)絡(luò) 347
16.1.3  添加DAG同步 347
16.1.4  使用CeleryExecutor擴展 348
16.1.5  后續(xù)步驟 349
16.2  針對AWS的hook和operator 350
16.3  用例：使用AWS Athena進行無服務(wù)器的電影排名 351
16.3.1  用例概要 352
16.3.2  設(shè)置資源 352
16.3.3  創(chuàng)建DAG 355
16.3.4  環(huán)境清理 360
16.4  本章小結(jié) 361
第17章  在Azure中使用Airflow 363
17.1  在Azure中部署Airflow 363
17.1.1  選擇服務(wù) 363
17.1.2  設(shè)計網(wǎng)絡(luò) 364
17.1.3  使用CeleryExecutor擴展 365
17.1.4  后續(xù)步驟 366
17.2  針對Azure設(shè)計的hook和operator 367
17.3  示例：在Azure上運行無服務(wù)器的電影推薦程序 367
17.3.1  示例概要 368
17.3.2  設(shè)定資源 368
17.3.3  創(chuàng)建DAG 372
17.3.4  環(huán)境清理 377
17.4  本章小結(jié) 378
第18章  在GCP中運行Airflow 379
18.1  在GCP中部署Airflow 379
18.1.1  選擇服務(wù) 379
18.1.2  使用Helm在GKE上
部署Airflow 381
18.1.3  與Google服務(wù)集成 383
18.1.4  設(shè)計網(wǎng)絡(luò) 385
18.1.5  通過CeleryExecutor擴展 386
18.2  針對GCP的hook和operator 388
18.3  用例：在GCP上運行無服務(wù)器的電影評級 392
18.3.1  上傳到GCS 392
18.3.2  將數(shù)據(jù)導(dǎo)入BigQuery 394
18.3.3  提取最高評分 396
18.4  本章小結(jié) 399
附錄A  運行示例代碼 401
附錄B  Airflow 1和Airflow 2中的包結(jié)構(gòu) 405
附錄C  Prometheus指標(biāo)映射 409