วิธีดึงข้อมูลจาก API สำหรับ Data Pipeline ด้วย Python

Application Programming Interfaces (APIs) ได้กลายเป็นแกนหลักของสถาปัตยกรรมข้อมูลสมัยใหม่ API เป็นช่องทางที่แอปพลิเคชันใช้สื่อสารและแลกเปลี่ยนข้อมูล ทำให้เป็นทรัพยากรที่มีค่าสำหรับการสร้าง Data Pipeline ที่แข็งแกร่งและมีความยืดหยุ่น ความสามารถในการดึงข้อมูลจาก API ได้อย่างมีประสิทธิภาพโดยใช้ภาษาที่หลากหลายอย่าง Python ถือเป็นทักษะพื้นฐานสำหรับ Data Engineer, Data Scientist หรือ Analyst ทุกคน บทความนี้จะเจาะลึกถึงความซับซ้อนของกระบวนการนี้ โดยให้คำแนะนำที่ครอบคลุมเกี่ยวกับวิธีการใช้ประโยชน์จากพลังของ API เพื่อขับเคลื่อน Data Pipeline ของคุณ

💡

ต้องการเครื่องมือทดสอบ API ที่ยอดเยี่ยมซึ่งสร้าง เอกสาร API ที่สวยงาม หรือไม่?

ต้องการแพลตฟอร์มแบบ All-in-One ที่รวมทุกอย่างเข้าด้วยกันเพื่อให้ทีม Developer ของคุณทำงานร่วมกันด้วย ประสิทธิภาพสูงสุด หรือไม่?

Apidog ตอบสนองความต้องการทั้งหมดของคุณ และ แทนที่ Postman ได้ในราคาที่ย่อมเยากว่ามาก!

button

บทบาทของ API ใน Data Pipeline

หัวใจหลักของ Data Pipeline คือชุดของกระบวนการอัตโนมัติที่ย้ายข้อมูลจากแหล่งหนึ่งไปยังปลายทางอีกแห่งหนึ่ง ขั้นตอนเริ่มต้นและอาจเป็นขั้นตอนที่สำคัญที่สุดของ Pipeline นี้คือการดึงข้อมูล แม้ว่าข้อมูลจะสามารถมาจากฐานข้อมูล ไฟล์ หรือแพลตฟอร์ม Streaming ได้ แต่ API ก็มีข้อได้เปรียบที่ไม่เหมือนใคร: การเข้าถึงข้อมูลแบบเรียลไทม์ มีความยืดหยุ่น และมักเป็นข้อมูลที่เป็นกรรมสิทธิ์จาก Web Service และแอปพลิเคชันที่หลากหลาย

ไม่ว่าจะเป็นการดึงข้อมูลทางการเงินจาก API ตลาดหุ้น การรวบรวมแนวโน้มโซเชียลมีเดียจาก API ของแพลตฟอร์ม หรือการเข้าถึงข้อมูลลูกค้าจาก API ของระบบ CRM ความสามารถในการดึงข้อมูลเหล่านี้ด้วยโปรแกรมถือเป็นพื้นฐาน Python ด้วยระบบนิเวศที่สมบูรณ์ของไลบรารีและไวยากรณ์ที่ตรงไปตรงมา ได้กลายเป็นภาษามาตรฐานสำหรับงานนี้ ความเรียบง่ายช่วยให้พัฒนาได้อย่างรวดเร็ว ในขณะที่ไลบรารีที่ทรงพลังมีเครื่องมือที่จำเป็นในการจัดการกับความซับซ้อนของการโต้ตอบกับ API

การเรียกใช้ API ครั้งแรกด้วย Python

การเริ่มต้นเข้าสู่การดึงข้อมูล API เริ่มต้นด้วยคำขอ HTTP อย่างง่าย ไลบรารี requests ใน Python เป็นมาตรฐานทองคำสำหรับวัตถุประสงค์นี้ มันช่วยลดความซับซ้อนของการสร้างคำขอ HTTP โดยมีอินเทอร์เฟซที่เรียบง่ายและสง่างาม

ในการเริ่มต้น คุณจะต้องติดตั้งไลบรารีก่อน:Python

pip install requests

เมื่อติดตั้งแล้ว คุณสามารถสร้างคำขอ GET ไปยัง API endpoint ได้ Endpoint คือ URL เฉพาะที่ให้ชุดข้อมูล สำหรับตัวอย่างนี้ เราจะใช้ JSONPlaceholder API ซึ่งเป็น REST API ออนไลน์ฟรีที่คุณสามารถใช้สำหรับการทดสอบและสร้างต้นแบบPython

import requests

response = requests.get('https://jsonplaceholder.typicode.com/posts/1')

# Check if the request was successful
if response.status_code == 200:
    data = response.json()
    print(data)
else:
    print(f"Failed to retrieve data: {response.status_code}")

ในโค้ดส่วนนี้ requests.get() ส่งคำขอ GET ไปยัง URL ที่ระบุ ออบเจกต์ response มีการตอบสนองของเซิร์ฟเวอร์ต่อคำขอของเรา แอตทริบิวต์ status_code บอกเราว่าคำขอสำเร็จหรือไม่ รหัสสถานะ 200 หมายถึงสำเร็จ เมธอด response.json() จะแยกวิเคราะห์เนื้อหา JSON ของการตอบสนองเป็น Python dictionary ทำให้ง่ายต่อการทำงานด้วย

การจัดการรูปแบบข้อมูลที่แตกต่างกัน

แม้ว่า JSON (JavaScript Object Notation) จะเป็นรูปแบบข้อมูลที่พบบ่อยที่สุดสำหรับ API แต่คุณอาจเจอรูปแบบอื่น ๆ เช่น XML (eXtensible Markup Language) ไลบรารี requests สามารถจัดการกับประเภทเนื้อหาที่แตกต่างกันได้ สำหรับ XML คุณอาจต้องใช้ไลบรารีเช่น xml.etree.ElementTree เพื่อแยกวิเคราะห์ข้อมูลPython

import requests
import xml.etree.ElementTree as ET

response = requests.get('URL_TO_XML_API')

if response.status_code == 200:
    root = ET.fromstring(response.content)
    # Now you can traverse the XML tree
    for child in root:
        print(child.tag, child.attrib)
else:
    print(f"Failed to retrieve data: {response.status_code}")

สิ่งสำคัญคือการตรวจสอบส่วนหัว Content-Type ของการตอบสนองเพื่อทำความเข้าใจรูปแบบของข้อมูลที่คุณได้รับ และใช้ไลบรารีการแยกวิเคราะห์ที่เหมาะสม

การเดินทางผ่านเขาวงกตของการยืนยันตัวตน API

API ส่วนใหญ่ต้องการการยืนยันตัวตนบางรูปแบบเพื่อระบุผู้ใช้และควบคุมการเข้าถึงข้อมูล สิ่งนี้สำคัญอย่างยิ่งสำหรับความปลอดภัยและการติดตามการใช้งาน API มีวิธีการยืนยันตัวตนทั่วไปหลายวิธี:

API Keys

นี่เป็นรูปแบบการยืนยันตัวตนที่ง่ายที่สุด ผู้ให้บริการ API จะให้คีย์เฉพาะแก่คุณ ซึ่งคุณต้องรวมไว้ในคำขอของคุณ คีย์นี้มักจะถูกส่งเป็นพารามิเตอร์การสอบถามใน URL หรือในส่วนหัวของคำขอPython

import requests

api_key = 'YOUR_API_KEY'
headers = {'Authorization': f'Bearer {api_key}'}

response = requests.get('https://api.example.com/data', headers=headers)

OAuth

OAuth (Open Authorization) เป็นมาตรฐานการยืนยันตัวตนที่ปลอดภัยและซับซ้อนกว่า ช่วยให้ผู้ใช้สามารถให้สิทธิ์แอปพลิเคชันของบุคคลที่สามเข้าถึงทรัพยากรของตนได้อย่างจำกัดโดยไม่ต้องแชร์ข้อมูลประจำตัว กระบวนการนี้มักจะเกี่ยวข้องกับการ Handshake หลายขั้นตอน โดยแอปพลิเคชันจะได้รับ Access Token ซึ่งจะถูกนำไปใช้ในการสร้างคำขอที่ได้รับการยืนยันตัวตน ไลบรารีเช่น requests-oauthlib สามารถช่วยลดความซับซ้อนของกระบวนการนี้ได้

Basic Authentication

วิธีนี้เกี่ยวข้องกับการส่งชื่อผู้ใช้และรหัสผ่านพร้อมกับแต่ละคำขอ ข้อมูลประจำตัวมักจะถูกเข้ารหัส Base64 และส่งในส่วนหัว Authorization ไลบรารี requests มีวิธีที่สะดวกในการจัดการกับสิ่งนี้:Python

from requests.auth import HTTPBasicAuth

response = requests.get('https://api.example.com/data', auth=HTTPBasicAuth('your_username', 'your_password'))

ศิลปะในการจัดการ Rate Limiting

เพื่อป้องกันการใช้งานที่ผิดวัตถุประสงค์และให้แน่ใจว่ามีการใช้งานที่เป็นธรรม API ส่วนใหญ่จะกำหนด Rate Limit ซึ่งจำกัดจำนวนคำขอที่ผู้ใช้สามารถทำได้ภายในระยะเวลาที่กำหนด การเกินขีดจำกัดนี้มักจะส่งผลให้ได้รับรหัสสถานะ 429 Too Many Requests สคริปต์การดึงข้อมูลที่แข็งแกร่งจะต้องจัดการกับขีดจำกัดเหล่านี้ได้อย่างราบรื่น

กลยุทธ์ทั่วไปคือการรวมช่วงเวลาการรอคอยไว้ในโค้ดของคุณ ไลบรารี time ใน Python เป็นเพื่อนของคุณที่นี่Python

import requests
import time

for i in range(100):
    response = requests.get('https://api.example.com/data')
    if response.status_code == 200:
        # Process the data
        pass
    elif response.status_code == 429:
        print("Rate limit exceeded. Waiting...")
        retry_after = int(response.headers.get('Retry-After', 10)) # Check for a 'Retry-After' header
        time.sleep(retry_after)
    else:
        print(f"An error occurred: {response.status_code}")
        break

ลูปอย่างง่ายนี้พยายามสร้างคำขอ หากเจอ Rate Limit จะตรวจสอบส่วนหัว Retry-After (ซึ่ง API บางตัวมีให้เพื่อระบุระยะเวลาที่ต้องรอ) จากนั้นจะหยุดการทำงานชั่วคราวก่อนที่จะลองอีกครั้ง

พิชิต Pagination: เรื่องราวที่ไม่มีวันจบสิ้น

เมื่อ API endpoint ส่งคืนชุดข้อมูลขนาดใหญ่ มักจะมีการ "แบ่งหน้า" (paginated) ซึ่งหมายความว่าข้อมูลจะถูกแบ่งออกเป็นหลายหน้า สคริปต์ของคุณจำเป็นต้องสามารถนำทางผ่านหน้าเหล่านี้เพื่อดึงข้อมูลทั้งหมด มีกลยุทธ์การแบ่งหน้าทั่วไปหลายวิธี:

Offset-Based Pagination

นี่เป็นหนึ่งในวิธีที่พบบ่อยที่สุด API จะมีพารามิเตอร์เช่น offset (หรือ page) และ limit (หรือ per_page) คุณจะเพิ่มค่า offset หรือ page ในคำขอถัดไปแต่ละครั้งเพื่อรับข้อมูลชุดถัดไปPython

import requests

base_url = 'https://api.example.com/data'
page = 1
all_data = []

while True:
    params = {'page': page, 'per_page': 100}
    response = requests.get(base_url, params=params)
    if response.status_code == 200:
        data = response.json()
        if not data: # No more data
            break
        all_data.extend(data)
        page += 1
    else:
        print(f"Failed to retrieve data: {response.status_code}")
        break

Cursor-Based Pagination

วิธีนี้ใช้ "Cursor" ซึ่งเป็นตัวชี้ไปยังรายการเฉพาะในชุดข้อมูล การตอบสนองของ API แต่ละครั้งจะมีฟิลด์ next_cursor หรือคล้ายกัน คุณใช้ Cursor นี้ในคำขอถัดไปเพื่อรับชุดข้อมูลถัดไป วิธีนี้โดยทั่วไปมีประสิทธิภาพมากกว่าสำหรับชุดข้อมูลขนาดใหญ่มากPython

import requests

base_url = 'https://api.example.com/data'
next_cursor = None
all_data = []

while True:
    params = {'cursor': next_cursor} if next_cursor else {}
    response = requests.get(base_url, params=params)
    if response.status_code == 200:
        data = response.json()
        all_data.extend(data['results'])
        next_cursor = data.get('next_cursor')
        if not next_cursor:
            break
    else:
        print(f"Failed to retrieve data: {response.status_code}")
        break

การจัดโครงสร้างและจัดเก็บข้อมูลที่ดึงมา

เมื่อคุณดึงข้อมูลจาก API ได้สำเร็จแล้ว ขั้นตอนต่อไปคือการจัดโครงสร้างและจัดเก็บข้อมูลในรูปแบบที่เหมาะสมสำหรับ Data Pipeline ของคุณ ข้อมูล JSON หรือ XML ดิบมักจะซ้อนกันและไม่เหมาะสำหรับการวิเคราะห์โดยตรงหรือการโหลดลงในฐานข้อมูลเชิงสัมพันธ์

ไลบรารี pandas เป็นเครื่องมือที่ขาดไม่ได้สำหรับงานนี้ มันมี DataFrame ซึ่งเป็นโครงสร้างข้อมูลสองมิติที่มีป้ายกำกับซึ่งเหมาะสำหรับข้อมูลในรูปแบบตารางPython

import pandas as pd

# Assuming 'all_data' is a list of dictionaries from the API
df = pd.DataFrame(all_data)

จากนั้นคุณสามารถดำเนินการแปลงข้อมูลต่างๆ บน DataFrame ได้ เช่น การเลือกคอลัมน์ที่ต้องการ การเปลี่ยนชื่อคอลัมน์ และการจัดการค่าที่หายไป

สำหรับการจัดเก็บเบื้องต้น คุณมีหลายทางเลือก:

CSV (Comma-Separated Values): รูปแบบที่เรียบง่ายและรองรับอย่างกว้างขวาง df.to_csv('data.csv', index=False)
JSON: มีประโยชน์หากคุณต้องการรักษาโครงสร้างที่ซ้อนกันของข้อมูลต้นฉบับ df.to_json('data.json', orient='records')
Parquet: รูปแบบการจัดเก็บแบบคอลัมน์ที่มีประสิทธิภาพสูงสำหรับงานวิเคราะห์ นี่มักเป็นตัวเลือกที่นิยมสำหรับ Data Lake df.to_parquet('data.parquet')
Database: สำหรับการจัดเก็บที่มีโครงสร้างและระยะยาวมากขึ้น คุณสามารถโหลดข้อมูลลงในฐานข้อมูล SQL หรือ NoSQL ได้โดยตรงโดยใช้ไลบรารีเช่น SQLAlchemy หรือ pymongo

การทำให้กระบวนการดึงข้อมูลเป็นอัตโนมัติ

Data Pipeline ไม่ใช่เรื่องที่ทำครั้งเดียวจบ คุณมักจะต้องดึงข้อมูลจาก API ตามกำหนดเวลาปกติ (เช่น รายวัน รายชั่วโมง) นี่คือที่มาของระบบอัตโนมัติ

คุณสามารถกำหนดเวลาให้สคริปต์ Python ของคุณทำงานตามช่วงเวลาที่กำหนดโดยใช้เครื่องมือเช่น:

Cron: ตัวกำหนดเวลางานตามเวลาในระบบปฏิบัติการที่เหมือน Unix
Windows Task Scheduler: เทียบเท่า Cron สำหรับ Windows
Airflow: แพลตฟอร์มที่ทรงพลังสำหรับการเขียนโปรแกรม กำหนดเวลา และตรวจสอบ Workflow Airflow เป็นตัวเลือกยอดนิยมสำหรับการสร้าง Data Pipeline ที่ซับซ้อน
Cloud-based Schedulers: บริการเช่น AWS Lambda with CloudWatch Events หรือ Google Cloud Functions with Cloud Scheduler ช่วยให้คุณสามารถรันสคริปต์ของคุณในสภาพแวดล้อม Serverless ได้

สรุป: การสร้างกระบวนการดึงข้อมูลที่ยืดหยุ่น

การดึงข้อมูลจาก API เป็นทักษะพื้นฐานสำหรับการสร้าง Data Pipeline ที่ทันสมัย แม้ว่าพื้นฐานของการสร้างคำขอ API จะตรงไปตรงมา แต่การสร้างกระบวนการดึงข้อมูลที่ยืดหยุ่นและพร้อมใช้งานจริงในระดับ Production ต้องพิจารณาอย่างรอบคอบเกี่ยวกับการยืนยันตัวตน Rate Limiting การแบ่งหน้า และการจัดการข้อผิดพลาด ด้วยการใช้ประโยชน์จากพลังของ Python และระบบนิเวศของไลบรารีที่สมบูรณ์ คุณสามารถเข้าถึงมหาสมุทรข้อมูลจำนวนมหาศาลที่มีให้ผ่าน API ได้อย่างมีประสิทธิภาพ และสร้าง Data Pipeline ที่แข็งแกร่งและเชื่อถือได้ การเดินทางจาก requests.get() อย่างง่าย ไปสู่สคริปต์การดึงข้อมูลที่เป็นอัตโนมัติและมีกำหนดเวลาอย่างเต็มรูปแบบ เป็นเครื่องยืนยันถึงพลังและความยืดหยุ่นของ Python ในโลกของ Data Engineering

💡

button