Tự động hóa tích hợp dữ liệu đa nguồn bằng Python và Cloud Storage

pip install requests boto3 pandas

import json
import boto3
import requests
from datetime import datetime

# Khởi tạo các hằng số cấu hình hệ thống
API_ENDPOINT = "https://api.example.com/v1/weather-data"
AWS_BUCKET_NAME = "my-company-data-lake"
AWS_REGION_NAME = "us-east-1"


def fetch_api_data(endpoint_url):
  """Hàm gửi yêu cầu HTTP GET để lấy dữ liệu thô từ API bên ngoài."""
  try:
    response = requests.get(endpoint_url, timeout=10)
    response.raise_for_status()
    print("Kết nối và lấy dữ liệu từ API thành công!")
    return response.json()
  except requests.exceptions.RequestException as error:
    print(f"Xảy ra lỗi trong quá trình gọi API: {error}")
    return None


def upload_to_cloud_storage(data_payload, bucket_name, file_name):
  """Hàm đóng gói dữ liệu JSON và tải lên dịch vụ Cloud Storage (AWS S3)."""
  s3_client = boto3.client("s3", region_name=AWS_REGION_NAME)
  try:
    formatted_json_data = json.dumps(data_payload, ensure_ascii=False, indent=4)
    s3_client.put_object(
        Bucket=bucket_name,
        Key=file_name,
        Body=formatted_json_data.encode("utf-8"),
    )
    print(f"Đã tải tệp lên Cloud Storage thành công với đường dẫn: {file_name}")
  except Exception as error:
    print(f"Lỗi khi truyền tải dữ liệu lên Cloud Storage: {error}")


def main_pipeline():
  """Hàm điều phối chính thực hiện toàn bộ luồng tích hợp dữ liệu."""
  raw_data = fetch_api_data(API_ENDPOINT)
  if raw_data:
    current_timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
    target_file_name = f"weather_reports/report_{current_timestamp}.json"
    upload_to_cloud_storage(raw_data, AWS_BUCKET_NAME, target_file_name)


if __name__ == "__main__":
  main_pipeline()

import io
import boto3
import pandas as pd
import requests

# Khai báo các hằng số kết nối
API_URL = "https://jsonplaceholder.typicode.com/users"
BUCKET_NAME = "my-company-data-lake"


def fetch_and_convert_users():
  """Lấy dữ liệu người dùng từ API và chuyển đổi thành DataFrame."""
  try:
    response = requests.get(API_URL, timeout=10)
    response.raise_for_status()
    users_data = response.json()

    # Chuyển đổi JSON thành Pandas DataFrame và chọn lọc các cột cần thiết
    df = pd.DataFrame(users_data)
    filtered_df = df[["id", "name", "email", "phone"]]
    return filtered_df
  except requests.exceptions.RequestException as error:
    print(f"Lỗi khi kết nối API: {error}")
    return None


def upload_dataframe_to_s3(dataframe):
  """Chuyển đổi DataFrame thành chuỗi CSV trong bộ nhớ và tải lên AWS S3."""
  if dataframe is not None:
    csv_buffer = io.StringIO()
    dataframe.to_csv(csv_buffer, index=False)

    s3_client = boto3.client("s3")
    try:
      s3_client.put_object(
          Bucket=BUCKET_NAME,
          Key="processed_data/users_list.csv",
          Body=csv_buffer.getvalue().encode("utf-8"),
      )
      print("Tải tệp CSV lên Cloud Storage thành công!")
    except Exception as error:
      print(f"Lỗi khi đẩy tệp lên S3: {error}")


if __name__ == "__main__":
  users_dataframe = fetch_and_convert_users()
  upload_dataframe_to_s3(users_dataframe)

import json
import boto3
import requests

# Khai báo các hằng số hệ thống
PRODUCT_API_URL = "https://fakestoreapi.com/products"
BUCKET_NAME = "my-company-data-lake"


def generate_product_report():
  """Lấy dữ liệu sản phẩm, lọc các mặt hàng giá cao và tính toán thống kê."""
  try:
    response = requests.get(PRODUCT_API_URL, timeout=10)
    response.raise_for_status()
    products_list = response.json()

    # Lọc các sản phẩm có giá trị lớn hơn 50.0 USD
    expensive_products = [
        item for item in products_list if item["price"] > 50.0
    ]

    total_items = len(expensive_products)
    average_price = (
        sum(item["price"] for item in expensive_products) / total_items
        if total_items > 0
        else 0
    )

    # Đóng gói dữ liệu thành một báo cáo hoàn chỉnh
    report_payload = {
        "total_expensive_products": total_items,
        "average_price": round(average_price, 2),
        "products": expensive_products,
    }
    return report_payload
  except requests.exceptions.RequestException as error:
    print(f"Lỗi khi gọi API sản phẩm: {error}")
    return None


def upload_json_report_to_s3(report_data):
  """Đẩy báo cáo JSON tổng hợp lên Cloud Storage."""
  if report_data:
    s3_client = boto3.client("s3")
    try:
      formatted_json = json.dumps(report_data, ensure_ascii=False, indent=4)
      s3_client.put_object(
          Bucket=BUCKET_NAME,
          Key="reports/expensive_products_summary.json",
          Body=formatted_json.encode("utf-8"),
      )
      print("Đẩy báo cáo tổng hợp lên Cloud Storage thành công!")
    except Exception as error:
      print(f"Lỗi khi tải báo cáo lên S3: {error}")


if __name__ == "__main__":
  summary_report = generate_product_report()
  upload_json_report_to_s3(summary_report)