สร้างไปป์ไลน์ Extract Transform Load โดยใช้ python และทำให้เป็นอัตโนมัติด้วยการไหลของอากาศ

ในโพสต์บล็อกนี้จะอธิบายวิธีสร้างไปป์ไลน์ ETL (แยก, แปลง, โหลด) อย่างง่ายโดยใช้ Python และทำให้กระบวนการเป็นอัตโนมัติผ่าน Apache airflow

คำชี้แจงปัญหา:

เราจำเป็นต้องใช้ API ของ Spotify เพื่ออ่านข้อมูลและทำการแปลงขั้นพื้นฐาน และในที่สุดการตรวจสอบคุณภาพข้อมูลจะโหลดข้อมูลที่ดึงมาไปยัง PostgreSQL DB และทำให้กระบวนการทั้งหมดเป็นไปโดยอัตโนมัติผ่านการไหลเวียนของอากาศ เวลาโดยประมาณ: [4–7 ชั่วโมง]

Tech Stack / Skill ที่ใช้:

1. หลาม
2. API ของ
3. นักเทียบท่า
4. การไหลของอากาศ
5. PostgreSQL

1. ความรู้เกี่ยวกับ API
2. ทำความเข้าใจว่านักเทียบท่าและนักเทียบท่าเขียนอะไร
3. Python ระดับกลางและ SQL
4. ความเข้าใจพื้นฐานของ Airflow สิ่งนี้จะช่วยได้

1. ทำความเข้าใจวิธีการโต้ตอบกับ API เพื่อดึงข้อมูล
2. จัดการ Dataframe ในแพนด้า
3. การตั้งค่า Airflow และ PostgreSQL ผ่าน Docker-Compose
4. เรียนรู้การสร้าง DAG ใน Airflow

โครงการนี้เป็นโครงการที่เป็นมิตรกับผู้เริ่มต้นเพื่อเริ่มต้นด้วยการสร้างไปป์ไลน์แบบง่ายๆ และระบบอัตโนมัติผ่านการไหลเวียนของอากาศ อันดับแรก เราจะมุ่งเน้นไปที่การสร้างไปป์ไลน์ทั้งหมด แล้วจึงขยายโครงการโดยรวมเข้ากับ Airflow

การสร้าง ETL ไปป์ไลน์:

ชุดข้อมูล:ในโปรเจกต์นี้ เราใช้ API ของ Spotify ดังนั้นโปรดดำเนินการต่อและสร้างบัญชีสำหรับตัวคุณเอง หลังจากสร้างบัญชีแล้วให้ไปที่หน้านี้ ตอนนี้คุณจะสามารถเห็นไอคอนรับโทเค็น คลิกที่นั้น และเลือกผู้ใช้ที่เพิ่งเล่น และคลิกรับโทเค็น

คุณจะเห็นโทเค็นของคุณในลักษณะนี้

ตอนนี้เป็นขั้นตอนในการรับโทเค็น คุณอาจต้องสร้างสิ่งนี้บ่อยครั้งเนื่องจากหมดอายุหลังจากผ่านไประยะหนึ่ง

Extract.py

เรากำลังใช้โทเค็นนี้เพื่อแยกข้อมูลจาก Spotify เรากำลังสร้างฟังก์ชัน return_dataframe() รหัสหลามด้านล่างอธิบายวิธีที่เราแยกข้อมูล API และแปลงเป็น Dataframe

Transform.py

ที่นี่เรากำลังส่งออกไฟล์ Extract เพื่อรับข้อมูล

def Data_Quality(load_df):ใช้เพื่อตรวจสอบกรอบข้อมูลที่ว่างเปล่า บังคับใช้ข้อจำกัดเฉพาะ ตรวจสอบค่า Null เนื่องจากข้อมูลเหล่านี้อาจทำให้ฐานข้อมูลของเราเสียหายได้ สิ่งสำคัญคือเราต้องบังคับใช้การตรวจสอบคุณภาพข้อมูลเหล่านี้

def Transform_df(load_df):ตอนนี้เรากำลังเขียนตรรกะบางอย่างตามความต้องการของเราที่นี่ เราต้องการรู้จักศิลปินคนโปรดของเรา ดังนั้นเราจึงจัดกลุ่มเพลงที่ศิลปินคนนั้นฟัง หมายเหตุ: ขั้นตอนนี้ไม่จำเป็น คุณสามารถนำไปใช้หรือตรรกะอื่นๆ ได้หากต้องการ แต่ให้แน่ใจว่าคุณบังคับใช้ข้อจำกัดหลัก

Load.py

ในขั้นตอนการโหลด เราใช้ sqlalchemy และ SQLite เพื่อโหลดข้อมูลของเราลงในฐานข้อมูลและบันทึกไฟล์ในไดเรกทอรีโครงการของเรา

ในที่สุด เราก็ดำเนินการไปป์ไลน์ ETL สำเร็จ โครงสร้างของโฟลเดอร์โครงการควรมีลักษณะดังนี้ (ภายในโฟลเดอร์โครงการเรามี 3 ไฟล์)

E:\DE\PROJECTS\SPOTIFY_ETL\SPOTIFY_ETL
│   Extract.py
│   Load.py
│   my_played_tracks.sqlite
│   spotify_etl.py
│   Transform.py
└───

ตอนนี้เราจะทำให้กระบวนการนี้เป็นไปโดยอัตโนมัติโดยใช้ Airflow

ทำงานอัตโนมัติผ่าน Airflow

สำหรับผู้ที่มาไกลถึงจุดนี้ ฉันขอขอบคุณความพยายามของคุณ แต่ต่อจากนี้จะยุ่งยากเล็กน้อย ดังนั้นฉันกำลังพูดถึงประเด็นสำคัญบางประการด้านล่าง

1. เราได้ทำ ETL เสร็จแล้วและนี่เป็นโครงการขนาดเล็กดังนั้นจึงบันทึกงานไว้
2. ตอนนี้เราจะขยายสิ่งนี้ด้วยการไหลของอากาศโดยใช้นักเทียบท่า
3. ทำไมต้องนักเทียบท่า? เราใช้นักเทียบท่าเพราะมันง่ายกว่าในการติดตั้งและบำรุงรักษา และมันไม่ขึ้นกับระบบปฏิบัติการ
4. จะตั้งค่าการไหลของอากาศโดยใช้ Docker ได้อย่างไร? ทำตามคำแนะนำที่ให้ไว้ในบล็อกนี้
5. คุณต้องเปลี่ยนไฟล์ Yaml เพียงอย่างเดียวจากหลักเกณฑ์ข้างต้น โปรดดูที่นี่
6. หลังจากตั้งค่าการไหลเวียนของอากาศแล้ว ให้วาง dags ของคุณไว้ในโฟลเดอร์ dags
7. หลังจากนักเทียบท่าเปิดขึ้นคุณจะเห็นบริการ 4 รายการทำงานอยู่

C:\USERS\SIDHA\DOCKER\AIRFLOW
│   docker-compose.yml
├───dags
│   │   YOUR_DAGS.py
├───logs
├───plugins
└───scripts

ได้เวลาสร้าง Dag ที่จำเป็นสำหรับโครงการของเราแล้ว แต่ก่อนที่จะข้ามไปที่ DAG ให้เราเข้าใจว่า dag คืออะไร DAG ย่อมาจาก Directed Acyclic Graph ซึ่งเป็นชุดของงานที่กำหนดไว้ตามลำดับการดำเนินการ

ดังนั้นในความคิดของเรา เราต้องสร้างงานเพื่อให้งานของเราสำเร็จลุล่วง เพื่อให้ง่ายฉันจะใช้งานสองอย่างคืองานหนึ่งสร้างตาราง Postgres และอีกงานหนึ่งเพื่อโหลดข้อมูลลงในตาราง dag ของเราจะมีลักษณะดังนี้

spotify_etl.py

ในไฟล์ Python นี้จะเขียนตรรกะเพื่อดึงข้อมูลจาก API → Do Quality Checks → Transform Data

1. เมื่อวาน = วันนี้ — datetime.timedelta(days=1) → กำหนดจำนวนวันที่คุณต้องการข้อมูล เปลี่ยนแปลงได้ตามต้องการ เนื่องจากงานของเราคือโหลดรายวัน ฉันตั้งไว้ที่ 1
2. def spotify_etl() → ฟังก์ชันหลักที่ส่งคืน Data Frame ไปยังไฟล์ DAG python
3. ต้องวางไฟล์นี้ไว้ในโฟลเดอร์ dags

นี่เป็นส่วนที่สำคัญที่สุดที่คุณต้องใส่ใจ ขั้นแรก เรียนรู้พื้นฐานเกี่ยวกับการไหลของอากาศ DAG ที่นี่อาจใช้เวลาประมาณ 15 นาที หรือคุณสามารถค้นหาได้ใน youtube หลังจากพื้นฐานแล้ว โปรดปฏิบัติตามคำแนะนำด้านล่าง

1. จาก airflow.operators.python_operator นำเข้า PythonOperator → เรากำลังใช้ตัวดำเนินการ python เพื่อดำเนินการฟังก์ชั่น python เช่นการแทรก DataFrame ลงในตาราง
2. จาก airflow.providers.postgres.operators.postgres นำเข้า PostgresOperator → เรากำลังใช้ตัวดำเนินการ Postgres เพื่อสร้างตารางในฐานข้อมูล Postgres ของเรา
3. จากการไหลเวียนของอากาศ hooks.base_hook นำเข้า BaseHook → hook เป็นนามธรรมของ API เฉพาะที่อนุญาตให้ Airflow โต้ตอบกับระบบภายนอก Hooks สร้างขึ้นในโอเปอเรเตอร์หลายตัว แต่ก็สามารถใช้ได้โดยตรงในโค้ด DAG เรากำลังใช้ hook ที่นี่เพื่อเชื่อมต่อฐานข้อมูล Postgres จากฟังก์ชัน python ของเรา
4. จาก spotify_etl นำเข้า spotify_etl → นำเข้าฟังก์ชัน spotify_etl จาก spotify_etl.py

การตั้งค่าอาร์กิวเมนต์เริ่มต้นและช่วงเวลา เราสามารถเปลี่ยนวันที่เริ่มต้นช่วงเวลาตามความต้องการของเรา

ทำความเข้าใจเกี่ยวกับการเชื่อมต่อและงานของ Postgres

1. conn = BaseHook.get_connection('[Your Connection ID]') → เชื่อมต่อกับ Postgres DB ของคุณ
2. df.to_sql('[Your Table Name]', engine, if_exists='replace') → โหลด DF ไปที่ตาราง
3. create_table >> run_etl → การกำหนดโฟลว์ของงาน

ไปที่ Airflow UI แล้วคลิกการเชื่อมต่อ

จากนั้นสร้างการเชื่อมต่อ

1. รหัสการเชื่อมต่อ: postgre_sql จะเป็นรหัสที่เราใช้ในรหัสของเรา
2. ประเภทการเชื่อมต่อ: Postgres
3. โฮสต์: Postgres
4. Schema: spotify_db คุณสามารถใช้ชื่อของคุณเองได้ แต่อย่าลืมตรวจสอบในฐานข้อมูลที่ถูกต้องเมื่อทำการตรวจสอบความถูกต้อง
5. เข้าสู่ระบบ:การไหลของอากาศ
6. รหัสผ่าน:การไหลของอากาศ
7. พอร์ต: 5432

การปรับใช้:

ตรวจสอบ Dag ของคุณในส่วน Dags

หลังจากเปิดใช้งานแล้วให้รัน dag โดยทริกเกอร์

หลังจากเสร็จสิ้นการตรวจสอบบันทึก

ตอนนี้เราจะตรวจสอบโดยเชื่อมต่อกับฐานข้อมูล Postgres เปิดเทอร์มินัลแล้วดำเนินการ: pgcli -h localhost -p 5432 -u airflow -d spotify_db มันจะถามรหัสผ่าน ป้อนกระแสลม ที่นี่spotify_dbคือชื่อฐานข้อมูลหรือสคีมาการเชื่อมต่อของเรา

พิมพ์ \d เพื่อดูตาราง

เราสามารถเห็นตารางของเราถูกสร้างขึ้น และตอนนี้มาตรวจสอบข้อมูลกัน

ในที่สุดเราก็ทำสำเร็จ :)

ตอนนี้ความต้องการของเราคือทำให้กระบวนการทั้งหมดนี้เป็นไปโดยอัตโนมัติเพื่อจุดประสงค์ในการสาธิต ให้ฉันเปลี่ยนช่วงเวลาการไหลของอากาศเป็น3 นาทีและฟังเพลงบ้าง

ขณะนี้การประทับเวลาปัจจุบันแสดงอยู่ด้านล่าง เรามาตรวจสอบกันหลังจากนี้สักครู่

ไชโย เราจัดให้แล้ว

เพลงใหม่ที่ฉันเพิ่งฟังได้รับการอัปโหลดไปยังฐานข้อมูลของเราโดยอัตโนมัติ

ตรวจสอบบันทึกและรายละเอียดของงานกันเถอะ มันเป็นงานที่กำหนดเวลาไว้

บทสรุป:

หมายเหตุ:เราอาจต้องเปลี่ยนโทเค็นในไฟล์ ETL python เนื่องจากโทเค็นจะหมดอายุหลังจากผ่านไประยะหนึ่ง มีข้อจำกัดบางประการสำหรับโครงการนี้ พวกเขาสามารถเอาชนะได้โดยใช้โทเค็นการรีเฟรชเพื่อต่ออายุโทเค็นโดยอัตโนมัติ และเราสามารถตั้งค่าการไหลเวียนของอากาศในบริการคลาวด์ให้ทำงานตลอด 24/7 และเลือกข้อมูลวันละครั้ง ทำให้เป็นโหลดรายวัน แต่เพื่อให้ เป็นมิตรกับผู้เริ่มต้นนี้ ฉันยังไม่ได้ครอบคลุมถึงสิ่งเหล่านั้น ฉันจะปล่อยให้สิ่งเหล่านั้นเป็นสิ่งที่ต้องทำ

Github Repo:https://github.com/sidharth1805/Spotify_etl. ฉันหวังว่าคุณจะสนุกกับโครงการแนะนำมาก ฉันค่อนข้างแน่ใจว่าคุณจะต้องเจอปัญหามากมายในขณะที่ทำโปรเจ็กต์ Stack overflow คือเพื่อนที่ดีที่สุดของเรา และอย่าลังเลที่จะติดต่อกับฉันทางLinkedInหากมีคำถามเพิ่มเติม ติดตามฉันบนสื่อเพื่อเรียนรู้เพิ่มเติมเกี่ยวกับวิศวกรรมข้อมูล