Upgrade to Pro
— share decks privately, control downloads, hide ads and more …
Speaker Deck
Features
Speaker Deck
PRO
Sign in
Sign up for free
Search
Search
Airflowのチュートリアルやってみた
Search
iwamot
June 30, 2023
Technology
0
160
Airflowのチュートリアルやってみた
2023-06-30
ENECHANGE Tech Talk(社内勉強会)
iwamot
June 30, 2023
Tweet
Share
More Decks by iwamot
See All by iwamot
ECS on FargateへのSeekable OCI導入レポート
iwamot
0
350
サービスクォータ、ちゃんと監視してますか?
iwamot
0
670
AWS Well-Architected Toolで信頼性をレビューした結果
iwamot
0
340
テックカンパニーとしてのENECHANGEの中期目標
iwamot
0
58
AIの活用状況と今後の展望
iwamot
0
49
ENECHANGE社のAWSセキュリティ改革ストーリー
iwamot
0
120
ALBのアクセスログをAthenaで分析してSLOをゆるく決める
iwamot
2
1.1k
ChatGPT in SlackでAI Slackbotを楽しく運用する
iwamot
1
350
Function callingによるSlackbotの機能拡張
iwamot
2
170
Other Decks in Technology
See All in Technology
グイグイ系QAエンジニアでやっていくよ!
____rina____
0
720
回り回って効いてくる副次的効果としての技術広報/techpr
nishiuma
1
170
多言語化対応における TypeScript の型定義を通して開発のしやすさについて考えた / TSKaigi TypeScript Multilingualization
nabeliwo
2
380
シンプルなHITL機械学習と様々なタスクにおけるHITL機械学習
naohachi89
0
310
Google Cloud Next '24 Recap in ZOZO AIにより変わる開発 運用/Development and operation changed by AI
gachimuchiengineer
0
190
AI JIMY - 登壇(インストール編)
hanacchi
0
150
使われないものを作るな!出口から作るデータ分析基盤 / Data Platform Development Starting from the User Needs
amaotone
16
4k
LLM評価の落とし穴~開発者目線で気をつけるポイント~
rishigami
10
3.1k
AWS Observability 関連最新アップデート
o11yfes2023
0
100
パスワードを保存しますか?
hanacchi
0
240
Zero Data Loss Autonomous Recovery Service サービス概要
oracle4engineer
PRO
0
2k
TDD - Test Driven Drupal
opdavies
0
3k
Featured
See All Featured
Ruby is Unlike a Banana
tanoku
96
10k
Clear Off the Table
cherdarchuk
85
310k
RailsConf 2023
tenderlove
9
580
ピンチをチャンスに:未来をつくるプロダクトロードマップ #pmconf2020
aki_iinuma
81
44k
[RailsConf 2023] Rails as a piece of cake
palkan
28
4k
For a Future-Friendly Web
brad_frost
172
9k
Facilitating Awesome Meetings
lara
43
5.6k
The Art of Programming - Codeland 2020
erikaheidi
43
12k
Building an army of robots
kneath
300
41k
Automating Front-end Workflow
addyosmani
1357
200k
YesSQL, Process and Tooling at Scale
rocio
165
13k
Six Lessons from altMBA
skipperchong
22
3k
Transcript
Airflowのチュートリアルやってみた 2023-06-30 ENECHANGE Tech Talk(社内勉強会) CTO室 岩本隆史
Airflow案件に関わることに
よい機会なのでMWAAを試したい https://aws.amazon.com/jp/managed-workflows-for-apache-airflow/
チュートリアルをやってみよう https://docs.aws.amazon.com/mwaa/latest/userguide/quick-start.html
めっちゃ時間かかった… https://docs.aws.amazon.com/mwaa/latest/userguide/quick-start.html#quick-start- createstack
Dockerだと数分で構築完了 curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.6.2/docker-compose.yaml' mkdir -p ./dags ./logs ./plugins ./config
echo -e "AIRFLOW_UID=$(id -u)" > .env docker compose up airflow-init docker compose up https://airflow.apache.org/docs/apache-airflow/stable/howto/docker-compose/
サンプルDAGも豊富
tutorial DAGを実行
成功
3つのタスク
タスク1=日付の出力 t1 = BashOperator( task_id="print_date", bash_command="date", ) [2023-06-22, 06:52:22 UTC]
{subprocess.py:75} INFO - Running command: ['/bin/bash', '-c', 'date'] [2023-06-22, 06:52:22 UTC] {subprocess.py:86} INFO - Output: [2023-06-22, 06:52:22 UTC] {subprocess.py:93} INFO - Thu Jun 22 06:52:22 UTC 2023 [2023-06-22, 06:52:22 UTC] {subprocess.py:97} INFO - Command exited with return code 0
タスク2=スリープ t2 = BashOperator( task_id="sleep", depends_on_past=False, bash_command="sleep 5", retries=3, )
[2023-06-22, 06:52:25 UTC] {subprocess.py:75} INFO - Running command: ['/bin/bash', '-c', 'sleep 5'] [2023-06-22, 06:52:25 UTC] {subprocess.py:86} INFO - Output: [2023-06-22, 06:52:30 UTC] {subprocess.py:97} INFO - Command exited with return code 0
タスク3=テンプレートの利用 templated_command = dedent( """ {% for i in range(5)
%} echo "{{ ds }}" echo "{{ macros.ds_add(ds, 7)}}" {% endfor %} """ ) t3 = BashOperator( task_id="templated", depends_on_past=False, bash_command=templated_command, )
10個のechoにレンダリング echo "2023-06-22" echo "2023-06-29" echo "2023-06-22" echo "2023-06-29" echo
"2023-06-22" echo "2023-06-29" echo "2023-06-22" echo "2023-06-29" echo "2023-06-22" echo "2023-06-29"
10個の日付が出力 [2023-06-22, 06:52:25 UTC] {subprocess.py:86} INFO - Output: [2023-06-22, 06:52:25
UTC] {subprocess.py:93} INFO - 2023-06-22 [2023-06-22, 06:52:25 UTC] {subprocess.py:93} INFO - 2023-06-29 [2023-06-22, 06:52:25 UTC] {subprocess.py:93} INFO - 2023-06-22 [2023-06-22, 06:52:25 UTC] {subprocess.py:93} INFO - 2023-06-29 [2023-06-22, 06:52:25 UTC] {subprocess.py:93} INFO - 2023-06-22 [2023-06-22, 06:52:25 UTC] {subprocess.py:93} INFO - 2023-06-29 [2023-06-22, 06:52:25 UTC] {subprocess.py:93} INFO - 2023-06-22 [2023-06-22, 06:52:25 UTC] {subprocess.py:93} INFO - 2023-06-29 [2023-06-22, 06:52:25 UTC] {subprocess.py:93} INFO - 2023-06-22 [2023-06-22, 06:52:25 UTC] {subprocess.py:93} INFO - 2023-06-29 [2023-06-22, 06:52:25 UTC] {subprocess.py:97} INFO - Command exited with return code 0
タスク依存関係は演算子で指定 t1 >> [t2, t3]
別のチュートリアルも実行
Extract @task() def extract(): data_string = '{"1001": 301.27, "1002": 433.21,
"1003": 502.22}' order_data_dict = json.loads(data_string) return order_data_dict Key Value return_value {'1001': 301.27, '1002': 433.21, '1003': 502.22}
Transform @task(multiple_outputs=True) def transform(order_data_dict: dict): total_order_value = 0 for value
in order_data_dict.values(): total_order_value += value return {"total_order_value": total_order_value} Key Value total_order_value 1236.7 return_value {'total_order_value': 1236.7}
Load @task() def load(total_order_value: float): print(f"Total order value is: {total_order_value:.2f}")
[2023-06-22, 07:55:00 UTC] {logging_mixin.py:149} INFO - Total order value is: 1236.70
タスク依存関係は自動解決 order_data = extract() order_summary = transform(order_data) load(order_summary["total_order_value"])
実はAirflow 2.0の新機能 @task def hello_name(name: str): print(f'Hello {name}!') hello_name('Airflow users')
Dockerで気軽に試そう