ETL 处理设计和性能

ETL 处理设计和性能

python
import csv
import psycopg2
# 数据抽取
def extract_data(file_path):
    data = []
    with open(file_path, 'r') as file:
        reader = csv.reader(file)
        for row in reader:
            data.append(row)
    return data
# 数据清洗
def clean_data(data):
    cleaned_data = []
    for row in data:
        cleaned_row = []
        for value in row:
            cleaned_value = value.strip()
            cleaned_row.append(cleaned_value)
        cleaned_data.append(cleaned_row)
    return cleaned_data
# 数据加载
def load_data(data):
    conn = psycopg2.connect(database='mydb', user='myuser', password='mypassword', host='localhost', port='5432')
    cursor = conn.cursor()
    for row in data:
        cursor.execute("INSERT INTO mytable (column1, column2, column3) VALUES (%s, %s, %s)", row)
    conn.commit()
    cursor.close()
    conn.close()
# 主函数
def main():
    file_path = 'data.csv'
    data = extract_data(file_path)
    cleaned_data = clean_data(data)
    load_data(cleaned_data)
if __name__ == '__main__':
    main()