Cách Trích Xuất Dữ Liệu Từ API Cho Data Pipeline Bằng Python

Giao diện Lập trình Ứng dụng (API) đã nổi lên như là trụ cột của kiến trúc dữ liệu hiện đại. Chúng là các kênh mà qua đó các ứng dụng giao tiếp và trao đổi thông tin, biến chúng thành một nguồn tài nguyên vô giá để xây dựng các đường ống dữ liệu mạnh mẽ và linh hoạt. Khả năng trích xuất dữ liệu hiệu quả từ API bằng một ngôn ngữ linh hoạt như Python là một kỹ năng nền tảng cho bất kỳ kỹ sư dữ liệu, nhà khoa học dữ liệu hoặc nhà phân tích nào. Bài viết này sẽ đi sâu vào sự phức tạp của quá trình này, cung cấp một hướng dẫn toàn diện về cách khai thác sức mạnh của API để cung cấp năng lượng cho các đường ống dữ liệu của bạn.

💡

Bạn muốn một công cụ Kiểm thử API tuyệt vời có thể tạo ra Tài liệu API đẹp mắt?

Bạn muốn một nền tảng tích hợp, Tất cả trong Một để Nhóm Phát triển của bạn làm việc cùng nhau với năng suất tối đa?

Apidog đáp ứng mọi yêu cầu của bạn và thay thế Postman với mức giá phải chăng hơn nhiều!

button

Vai trò của API trong Đường ống Dữ liệu

Về cốt lõi, đường ống dữ liệu là một chuỗi các quy trình tự động di chuyển dữ liệu từ nguồn đến đích. Giai đoạn ban đầu và có lẽ quan trọng nhất của đường ống này là trích xuất dữ liệu. Mặc dù dữ liệu có thể được lấy từ cơ sở dữ liệu, tệp hoặc nền tảng truyền trực tuyến, API mang lại một lợi thế độc đáo: quyền truy cập vào dữ liệu thời gian thực, động và thường là độc quyền từ một loạt các dịch vụ web và ứng dụng.

Cho dù đó là lấy dữ liệu tài chính từ API thị trường chứng khoán, thu thập xu hướng truyền thông xã hội từ API của một nền tảng hay truy cập thông tin khách hàng từ API của hệ thống CRM, khả năng truy xuất thông tin này một cách lập trình là điều cơ bản. Python, với hệ sinh thái thư viện phong phú và cú pháp đơn giản, đã trở thành ngôn ngữ mặc định cho nhiệm vụ này. Sự đơn giản của nó cho phép phát triển nhanh chóng, trong khi các thư viện mạnh mẽ của nó cung cấp các công cụ cần thiết để xử lý sự phức tạp của tương tác API.

Thực hiện Cuộc gọi API Đầu tiên của Bạn với Python

Hành trình trích xuất dữ liệu API bắt đầu bằng một yêu cầu HTTP đơn giản. Thư viện requests trong Python là tiêu chuẩn vàng cho mục đích này. Nó trừu tượng hóa sự phức tạp của việc thực hiện các yêu cầu HTTP, cung cấp một giao diện đơn giản và thanh lịch.

Để bắt đầu, trước tiên bạn cần cài đặt thư viện:Python

pip install requests

Sau khi cài đặt, bạn có thể thực hiện yêu cầu GET đến một điểm cuối API. Điểm cuối đơn giản là một URL cụ thể cung cấp một tập hợp dữ liệu. Đối với ví dụ này, hãy sử dụng JSONPlaceholder API, một API REST trực tuyến miễn phí mà bạn có thể sử dụng để kiểm thử và tạo mẫu.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}")

Trong đoạn mã này, requests.get() gửi một yêu cầu GET đến URL được chỉ định. Đối tượng response chứa phản hồi của máy chủ đối với yêu cầu của chúng ta. Thuộc tính status_code cho chúng ta biết liệu yêu cầu có thành công hay không. Mã trạng thái 200 cho biết thành công. Phương thức response.json() sau đó phân tích nội dung JSON của phản hồi thành một từ điển Python, giúp dễ dàng làm việc.

Xử lý các Định dạng Dữ liệu Khác nhau

Mặc dù JSON (JavaScript Object Notation) là định dạng dữ liệu phổ biến nhất cho API, bạn có thể gặp các định dạng khác, chẳng hạn như XML (eXtensible Markup Language). Thư viện requests có thể xử lý các loại nội dung khác nhau. Đối với XML, bạn có thể cần sử dụng một thư viện như xml.etree.ElementTree để phân tích dữ liệu.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}")

Điều quan trọng là kiểm tra tiêu đề Content-Type của phản hồi để hiểu định dạng dữ liệu bạn đang nhận và sử dụng thư viện phân tích phù hợp.

Điều hướng Mê cung Xác thực API

Hầu hết các API đều yêu cầu một số hình thức xác thực để nhận dạng người dùng và kiểm soát quyền truy cập vào dữ liệu. Điều này rất quan trọng đối với bảo mật và theo dõi việc sử dụng API. Có một số phương pháp xác thực phổ biến:

Khóa API (API Keys)

Đây là một trong những hình thức xác thực đơn giản nhất. Nhà cung cấp API cung cấp cho bạn một khóa duy nhất mà bạn phải bao gồm trong các yêu cầu của mình. Khóa này thường được truyền dưới dạng tham số truy vấn trong URL hoặc trong các tiêu đề yêu cầu.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) là một tiêu chuẩn xác thực an toàn và phức tạp hơn. Nó cho phép người dùng cấp cho các ứng dụng bên thứ ba quyền truy cập hạn chế vào tài nguyên của họ mà không cần chia sẻ thông tin đăng nhập của họ. Quá trình¹ thường bao gồm một bắt tay nhiều bước, trong đó ứng dụng lấy mã truy cập, sau đó được sử dụng để thực hiện các yêu cầu đã được xác thực. Các thư viện như requests-oauthlib có thể đơn giản hóa quá trình này.

Xác thực Cơ bản (Basic Authentication)

Phương pháp này liên quan đến việc gửi tên người dùng và mật khẩu cùng với mỗi yêu cầu. Thông tin đăng nhập thường được mã hóa Base64 và gửi trong tiêu đề Authorization. Thư viện requests có một cách thuận tiện để xử lý việc này:Python

from requests.auth import HTTPBasicAuth

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

Nghệ thuật Xử lý Giới hạn Tốc độ (Rate Limiting)

Để ngăn chặn việc lạm dụng và đảm bảo sử dụng công bằng, hầu hết các API đều áp đặt giới hạn tốc độ, giới hạn số lượng yêu cầu mà người dùng có thể thực hiện trong một khoảng thời gian nhất định. Vượt quá giới hạn này thường sẽ dẫn đến mã trạng thái 429 Too Many Requests. Một tập lệnh trích xuất dữ liệu mạnh mẽ phải xử lý các giới hạn này một cách khéo léo.

Một chiến lược phổ biến là kết hợp một khoảng thời gian chờ trong mã của bạn. Thư viện time trong Python là người bạn của bạn ở đây.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

Vòng lặp đơn giản này cố gắng thực hiện các yêu cầu. Nếu gặp giới hạn tốc độ, nó sẽ kiểm tra tiêu đề Retry-After (mà một số API cung cấp để chỉ ra thời gian chờ bao lâu) và sau đó tạm dừng thực thi trước khi thử lại.

Chinh phục Phân trang (Pagination): Câu chuyện Không hồi kết

Khi một điểm cuối API trả về một tập dữ liệu lớn, nó thường được "phân trang", nghĩa là dữ liệu được chia thành nhiều trang. Tập lệnh của bạn cần có khả năng điều hướng qua các trang này để trích xuất tất cả dữ liệu. Có một số chiến lược phân trang phổ biến:

Phân trang Dựa trên Offset

Đây là một trong những phương pháp phổ biến nhất. API sẽ có các tham số như offset (hoặc page) và limit (hoặc per_page). Bạn tăng offset hoặc số page trong mỗi yêu cầu tiếp theo để lấy phần dữ liệu tiếp theo.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

Phân trang Dựa trên Con trỏ (Cursor)

Phương pháp này sử dụng một "con trỏ", là một con trỏ đến một mục cụ thể trong tập dữ liệu. Mỗi phản hồi API sẽ bao gồm trường next_cursor hoặc tương tự. Bạn sử dụng con trỏ này trong yêu cầu tiếp theo để lấy tập dữ liệu tiếp theo. Phương pháp này nói chung hiệu quả hơn đối với các tập dữ liệu rất lớn.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

Cấu trúc và Lưu trữ Dữ liệu Đã trích xuất

Sau khi bạn đã trích xuất dữ liệu thành công từ API, bước tiếp theo là cấu trúc và lưu trữ nó theo cách phù hợp với đường ống dữ liệu của bạn. Dữ liệu JSON hoặc XML thô thường bị lồng nhau và không lý tưởng để phân tích trực tiếp hoặc tải vào cơ sở dữ liệu quan hệ.

Thư viện pandas là một công cụ không thể thiếu cho nhiệm vụ này. Nó cung cấp DataFrame, một cấu trúc dữ liệu hai chiều có nhãn, hoàn hảo cho dữ liệu dạng bảng.Python

import pandas as pd

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

Sau đó, bạn có thể thực hiện các phép biến đổi khác nhau trên DataFrame, chẳng hạn như chọn các cột cụ thể, đổi tên cột và xử lý các giá trị bị thiếu.

Đối với lưu trữ ban đầu, bạn có một số tùy chọn:

CSV (Comma-Separated Values): Một định dạng đơn giản và được hỗ trợ rộng rãi. df.to_csv('data.csv', index=False)
JSON: Hữu ích nếu bạn muốn giữ nguyên cấu trúc lồng nhau của dữ liệu gốc. df.to_json('data.json', orient='records')
Parquet: Một định dạng lưu trữ theo cột rất hiệu quả cho các tác vụ phân tích. Đây thường là lựa chọn ưu tiên cho các hồ dữ liệu (data lakes). df.to_parquet('data.parquet')
Cơ sở dữ liệu: Để lưu trữ có cấu trúc và dài hạn hơn, bạn có thể tải dữ liệu trực tiếp vào cơ sở dữ liệu SQL hoặc NoSQL bằng cách sử dụng các thư viện như SQLAlchemy hoặc pymongo.

Tự động hóa Quá trình Trích xuất

Đường ống dữ liệu không phải là một công việc chỉ làm một lần. Bạn sẽ thường cần trích xuất dữ liệu từ API theo lịch trình thường xuyên (ví dụ: hàng ngày, hàng giờ). Đây là lúc tự động hóa phát huy tác dụng.

Bạn có thể lên lịch cho các tập lệnh Python của mình chạy theo các khoảng thời gian cụ thể bằng cách sử dụng các công cụ như:

Cron: Một trình lên lịch công việc dựa trên thời gian trong các hệ điều hành giống Unix.
Windows Task Scheduler: Tương đương với Cron cho Windows.
Airflow: Một nền tảng mạnh mẽ để lập trình, lên lịch và giám sát quy trình làm việc. Airflow là một lựa chọn phổ biến để xây dựng các đường ống dữ liệu phức tạp.
Bộ lên lịch dựa trên đám mây (Cloud-based Schedulers): Các dịch vụ như AWS Lambda với CloudWatch Events hoặc Google Cloud Functions với Cloud Scheduler cho phép bạn chạy các tập lệnh của mình trong môi trường phi máy chủ (serverless).

Kết luận: Xây dựng một Quy trình Trích xuất Mạnh mẽ

Trích xuất dữ liệu từ API là một kỹ năng nền tảng để xây dựng các đường ống dữ liệu hiện đại. Mặc dù các kiến thức cơ bản về thực hiện yêu cầu API rất đơn giản, việc xây dựng một quy trình trích xuất mạnh mẽ và sẵn sàng cho sản xuất đòi hỏi phải xem xét cẩn thận về xác thực, giới hạn tốc độ, phân trang và xử lý lỗi. Bằng cách tận dụng sức mạnh của Python và hệ sinh thái thư viện phong phú của nó, bạn có thể khai thác hiệu quả đại dương dữ liệu rộng lớn có sẵn thông qua API và xây dựng các đường ống dữ liệu vừa mạnh mẽ vừa đáng tin cậy. Hành trình từ một requests.get() đơn giản đến một tập lệnh trích xuất dữ liệu hoàn toàn tự động và theo lịch trình là minh chứng cho sức mạnh và tính linh hoạt của Python trong thế giới kỹ thuật dữ liệu.

💡

button