一、实战场景

如何使用 Pandas 清洗的二手房数据并存储文件

二、知识点

python 基础语法

python 文件读写

pandas 数据清洗

三、菜鸟实战

清洗前的文件

读取源文件

    def do_clean_data(self):
        # 执行清洗

        clean_data_raw_file_path = self.fileManger.get_data_file_path(self.clean_data_raw_file)

        if not os.path.isfile(clean_data_raw_file_path):
            # 确认文件存在
            self.logger.error("需要清洗的文件不存在")
            print("需要清洗的文件不存在")
            return False

        # 存储清洗数据数组, 先把文件头部存储
        new_row_arr = []

        # 从清洗文件读取需要清洗的数据
        raw_df = pd.read_csv(clean_data_raw_file_path, encoding=self.encoding)

        for idx in tqdm.trange(len(raw_df.values)):
            row = raw_df.values[idx]
            if self.check_row_is_valid(row):
                # 检查 row 的合法性
                new_row = self.clean_row(row)
                new_row_arr.append(new_row)

        # 保存文件
        self.save_to_clean_file(new_row_arr)

对二手房数据进行清洗

from base_cleaner import BaseDataCleaner
import time
import platform


class Tao365Cleaner(BaseDataCleaner):
    # 逐行读取文件，进行数据清洗

    clean_data_raw_file = "tao365_detail.csv"   # 要执行数据清洗的文件
    clean_data_result_file = 'tao365_clean.csv'  # 数据清洗的结果文件
    clean_data_result_file_head = ['标题', '价格', '每平方价格', '小区', '地址', '房屋户型', '建筑面积', '所在楼层', '房屋朝向', '建筑年代', '建成年份', '原始房屋户型']  # 数据清洗的结果文件的头部信息

    def check_row_is_valid(self, raw_row):
        # 检查当前行是否有效
        area = '区' in raw_row[4]
        if area == False:
            return False
        year = '年' in raw_row[9]
        if year == False:
            return False
        return True

    def clean_row(self, raw_row):
        # 执行单行清洗
        # self.logger.info("当前清洗数据: ", raw_row)
        # print(raw_row)
        # 标题
        title = raw_row[0]
        # 价格
        price = self.get_price(raw_row[1])
        # 每平方价格
        avg_price_per_square_meter = self.get_avg_price_per_square_meter(raw_row[2])
        # 小区
        housing = self.get_community(raw_row[3])
        # 地址
        area = self.get_area(raw_row[4])
        # 房屋户型
        house_type = self.get_room_count(raw_row[5])
        house_type_2 = raw_row[5]
        # 建筑面积
        acreage = self.get_area_size(raw_row[6])
        # 所在楼层
        level = self.get_floor(raw_row[7])
        # 房屋朝向
        direction = raw_row[8]
        # 年代
        year = self.get_year(raw_row[9])
        # 建成年份
        time = self.get_house_age(year)
        # 数据转换
        new_row = [title, price, avg_price_per_square_meter, housing, area, house_type, acreage, level, direction, year, time, house_type_2]

        return new_row

    def get_price(self, row):
        # 售价
        price = row.replace('万', '')
        return int(float(price))

    def get_avg_price_per_square_meter(self, row):
        # 每平方米均价
        avg_price_per_square_meter = row.replace('元/m²', '')
        return avg_price_per_square_meter

    def get_area(self, row):
        # 获取属于哪个区
        area = row[:row.index('区')]
        return area + '区'

    def get_community(self, row):
        # 获取属于哪个小区
        return row

    def get_room_count(self, row):
        # 获取房型是几室
        room_count = row[:row.index('室')]
        return room_count

    def get_area_size(self, row):
        # 获取面积
        area_size = row.replace('㎡', '')
        return area_size

    def get_floor(self, row):
        # 获取所属楼层
        floor = row[row.index('/'):]
        return floor.replace('/', '').replace('层', '')

    def get_year(self, row):
        # 获取建筑年代
        return row.replace('年', '')

    def get_house_age(self, row):
        # 获取房屋年龄建筑年代
        localtime = time.localtime(time.time())
        year = int(localtime[0]) - int(row)
        return year

    def test(self):
        # 测试
        raw_row = ['娥眉新村 4层 110平米', '460.4万', '70769元/m²', '北极西村', '玄武区玄武门北极西村21号', '4室1厅2卫', '65㎡', '高楼层/3层', '南北', '2020年']
        print(raw_row)
        new_row = self.clean_row(raw_row)
        print(new_row)


if __name__ == '__main__':
    print("数据清洗开始")

    raw_row = ['娥眉新村 4层 110平米', '460.4万', '70769元/m²', '北极西村', '玄武区玄武门北极西村21号', '4室1厅2卫',
               '65㎡', '高楼层/3层', '南北', '2020年']

    cleaner = Tao365Cleaner()
    # cleaner.test()
    cleaner.do_clean_data()

    print("数据清洗完成")
    print("python 版本", platform.python_version())

清洗完成后保存到文件

    def save_to_clean_file(self, data_arr):
        # 保存到清洗的文件
        file_path = self.fileManger.get_data_file_path(self.clean_data_result_file)

        # 初始化数据
        frame = pd.DataFrame(data_arr)
        frame.columns = self.clean_data_result_file_head
        frame.to_csv(file_path, encoding=self.encoding, index=None)
        self.logger.debug("清洗文件保存完成")