บล็อกนี้ ยังเป็นเรื่องเกี่ยวกับ Airflow อยู่ อย่าเพิ่งเบื่อกันก่อนนะฮะ

จะเล่าถึงการทำ unit test เพื่อให้แน่ใจว่า DAG ของเรานั้นเขียนมาถูกต้องในระดับหนึ่งแล้ว สามารถรันได้ ใช้งานได้ ไม่ใช่ deploy แล้วเจ๊ง แบบนี้ก็ไม่โอเคเนอะ ในบล็อกนี้ เลยจะพูดถึง DAG integrity กันฮะ

ทำไปทำไม

อย่างที่เกริ่นไป การที่เราทำ unit test ของ DAG integrity ก็คือ การที่ DAG นั้นมีความสมบูรณ์ (integrity) อันนี้เป็นคำนึงที่หาแปลไทยยากเหมือนกันนะฮะ ไม่ค่อยตรงเสียทีเดียวแต่พอกล้อมแกล้มได้แหละ

คำว่า integrity ในที่นี้ ให้นึกถึงว่า source code ใน DAG ของเรานั้นมันใช้งานได้นะ รันผ่านได้นะ ไม่มี syntax error หรือ library error หรือ dependency conflict ใดๆ ซึ่งก็เป็นระดับเบื้องต้น และเราก็จะทำ test ในระดับ unit test กันฮะ

มาเริ่มกัน

DAGBAG

DAGBAG เป็น module นึงใน Airflow DAG ซึ่งมันจะเก็บรายชื่อ DAG ที่ compile ไว้แล้ว และมีข้อมูล metadata อีกด้วย รายละเอียดหาอ่านได้จากลิงก์นี้นะฮะ

เราจะใช้ประโยชน์จากมันกัน แบบ code ข้างล่างนี้ฮะ

พอเรา import DagBag และสร้าง object dagbag ในบรรทัดที่ 3 เสร็จแล้ว เราจะสามารถเรียกใช้  attributes .dags และ .import_errors ได้ มันจะแสดงรายการ DAG ที่ compile ผ่านและ compile ไม่ผ่านตามลำดับฮะ

ตรงนี้ คือเหมือนกับ command ที่เคยเล่าไปในบล็อกก่อนหน้านี้ ถ้าสนใจก็กดลิงก์ข้างล่างไปอ่านอีกรอบได้นะฮะ

Let’s try: Airflow 2

Airflow 2 comes with lots of improvements. Why not spend some times to get know this for easier batch job development?

bluebirz.netbluebirz

แล้วก็ใช้ร่วมกับ unittest

unittest เป็นอีกหนึ่ง package ที่ใช้ทดสอบโปรแกรมของเรานะฮะ เราก็จะเอามาใช้ร่วมกับ DagBag แบบรูปด้านล่างนี้

มีแบบ git file ด้วยฮะ จิ้มๆ

airflow-docker/dag_integrity.py at main · bluebirz/airflow-docker

Docker-compose for local airflow development. Contribute to bluebirz/airflow-docker development by creating an account on GitHub.

GitHubbluebirz

พอเราจะใช้ ก็รันคำสั่งนี้ฮะ มันจะเข้าไปเช็คว่าใน DagBag เนี่ย มี DAG ไหนบ้างที่ไม่เรียบร้อย อันไหนบ้างที่โอเคแล้ว

python tests/dag_integrity.py

ถ้าผ่านหมด จะเห็นแบบรูปนี้

แต่ถ้าไม่ผ่านล่ะ ก็จะเห็นสีแดงๆ แบบนี้ฮะ

เคสใช้งานจริง

อันนี้เป็นการสาธิตว่าเราจะตรวจสอบ DAG ได้อย่างไร แบบอัตโนมือนะฮะ แต่เวลาใช้งานจริง เราจะรันคำสั่งข้างบนใน CI/CD ของเราเองฮะ โดยจะรันก่อน step ที่จะ deploy ไปยัง server ปลายทาง ซึ่งถ้ามันรันไม่ได้ขึ้นมาจริงๆ จะได้ไม่เกิดปัญหาที่ server เราจะเห็นก่อนและแก้ไขทันฮะ

ซึ่งการเขียน CI/CD step ก็แล้วแต่ design และ tools ที่เราใช้เลยนะฮะ ไม่ว่าจะเป็น Github action, Bitbucket pipeline, Google Cloud Build หรืออื่นๆ เลือกเอาตามที่ถนัดได้เลยฮะ

หวังว่าจะมีประโยชน์ถ้วนหน้า บั๊กไม่มากล้ำกรายนะฮะ