+-------------+      +--------------+      +--------------+      +---------------+
| 數(shù)據(jù)采集    | ---> | 數(shù)據(jù)清洗     | ---> | 數(shù)據(jù)轉(zhuǎn)換     | ---> | 數(shù)據(jù)存儲(chǔ)      |
| (API 請(qǐng)求)  |      | (去除無(wú)效數(shù)據(jù)) |      | (結(jié)構(gòu)化數(shù)據(jù)) |      | (保存到數(shù)據(jù)庫(kù)) |
+-------------+      +--------------+      +--------------+      +---------------+

import requests
import pandas as pd
import sqlite3

# 數(shù)據(jù)采集 - 從 API 獲取數(shù)據(jù)
def fetch_data():
    url = "https://jsonplaceholder.typicode.com/posts"
    response = requests.get(url)
    if response.status_code == 200:
        data = response.json()
        return data
    else:
        raise Exception(f"Failed to fetch data: {response.status_code}")

# 調(diào)用數(shù)據(jù)采集函數(shù)
data = fetch_data()
print(f"獲取到的數(shù)據(jù)數(shù)量: {len(data)}")

# 數(shù)據(jù)清洗 - 使用 Pandas 對(duì)數(shù)據(jù)進(jìn)行清洗
def clean_data(data):
    df = pd.DataFrame(data)
    # 刪除包含空值的行
    df.dropna(inplace=True)
    return df

# 調(diào)用數(shù)據(jù)清洗函數(shù)
df_cleaned = clean_data(data)
print(f"清洗后的數(shù)據(jù): \n{df_cleaned.head()}")

# 數(shù)據(jù)轉(zhuǎn)換 - 處理并結(jié)構(gòu)化數(shù)據(jù)
def transform_data(df):
    # 只保留特定的列
    df_transformed = df[["userId", "id", "title", "body"]]
    # 重命名列以便更好理解
    df_transformed.rename(columns={"userId": "user_id", "id": "post_id"}, inplace=True)
    return df_transformed

# 調(diào)用數(shù)據(jù)轉(zhuǎn)換函數(shù)
df_transformed = transform_data(df_cleaned)
print(f"轉(zhuǎn)換后的數(shù)據(jù): \n{df_transformed.head()}")

# 數(shù)據(jù)存儲(chǔ) - 將數(shù)據(jù)保存到 SQLite 數(shù)據(jù)庫(kù)
def store_data(df):
    # 創(chuàng)建與 SQLite 數(shù)據(jù)庫(kù)的連接
    conn = sqlite3.connect("data_pipeline.db")
    # 將數(shù)據(jù)存儲(chǔ)到名為 'posts' 的表中
    df.to_sql("posts", conn, if_exists="replace", index=False)
    # 關(guān)閉數(shù)據(jù)庫(kù)連接
    conn.close()
    print("數(shù)據(jù)已成功存儲(chǔ)到數(shù)據(jù)庫(kù)中")

# 調(diào)用數(shù)據(jù)存儲(chǔ)函數(shù)
store_data(df_transformed)

import requests
import pandas as pd
import sqlite3

# 數(shù)據(jù)采集
def fetch_data():
    url = "https://jsonplaceholder.typicode.com/posts"
    response = requests.get(url)
    if response.status_code == 200:
        data = response.json()
        return data
    else:
        raise Exception(f"Failed to fetch data: {response.status_code}")

# 數(shù)據(jù)清洗
def clean_data(data):
    df = pd.DataFrame(data)
    df.dropna(inplace=True)
    return df

# 數(shù)據(jù)轉(zhuǎn)換
def transform_data(df):
    df_transformed = df[["userId", "id", "title", "body"]]
    df_transformed.rename(columns={"userId": "user_id", "id": "post_id"}, inplace=True)
    return df_transformed

# 數(shù)據(jù)存儲(chǔ)
def store_data(df):
    conn = sqlite3.connect("data_pipeline.db")
    df.to_sql("posts", conn, if_exists="replace", index=False)
    conn.close()
    print("數(shù)據(jù)已成功存儲(chǔ)到數(shù)據(jù)庫(kù)中")

# 構(gòu)建數(shù)據(jù)處理流水線
def data_pipeline():
    data = fetch_data()
    df_cleaned = clean_data(data)
    df_transformed = transform_data(df_cleaned)
    store_data(df_transformed)

# 運(yùn)行數(shù)據(jù)處理流水線
data_pipeline()