Python构建一个简单的数据处理流水线

+-------------+      +--------------+      +--------------+      +---------------+
| 数据采集    | ---> | 数据清洗     | ---> | 数据转换     | ---> | 数据存储      |
| (API 请求)  |      | (去除无效数据) |      | (结构化数据) |      | (保存到数据库) |
+-------------+      +--------------+      +--------------+      +---------------+

import requests
import pandas as pd
import sqlite3

# 数据采集 - 从 API 获取数据
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}")

# 调用数据采集函数
data = fetch_data()
print(f"获取到的数据数量: {len(data)}")

# 数据清洗 - 使用 Pandas 对数据进行清洗
def clean_data(data):
    df = pd.DataFrame(data)
    # 删除包含空值的行
    df.dropna(inplace=True)
    return df

# 调用数据清洗函数
df_cleaned = clean_data(data)
print(f"清洗后的数据: \n{df_cleaned.head()}")

# 数据转换 - 处理并结构化数据
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

# 调用数据转换函数
df_transformed = transform_data(df_cleaned)
print(f"转换后的数据: \n{df_transformed.head()}")

# 数据存储 - 将数据保存到 SQLite 数据库
def store_data(df):
    # 创建与 SQLite 数据库的连接
    conn = sqlite3.connect("data_pipeline.db")
    # 将数据存储到名为 'posts' 的表中
    df.to_sql("posts", conn, if_exists="replace", index=False)
    # 关闭数据库连接
    conn.close()
    print("数据已成功存储到数据库中")

# 调用数据存储函数
store_data(df_transformed)

import requests
import pandas as pd
import sqlite3

# 数据采集
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}")

# 数据清洗
def clean_data(data):
    df = pd.DataFrame(data)
    df.dropna(inplace=True)
    return df

# 数据转换
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

# 数据存储
def store_data(df):
    conn = sqlite3.connect("data_pipeline.db")
    df.to_sql("posts", conn, if_exists="replace", index=False)
    conn.close()
    print("数据已成功存储到数据库中")

# 构建数据处理流水线
def data_pipeline():
    data = fetch_data()
    df_cleaned = clean_data(data)
    df_transformed = transform_data(df_cleaned)
    store_data(df_transformed)

# 运行数据处理流水线
data_pipeline()