数据ETL-基于Python的数据预处理案例
作者：互联网
目的：
从数据收集，数据预处理，数据简单的汇总统计，以及后续的数据说明做一个简单的示例
本分析不涉及具体姓名的数据，做相应的匿名化处理，所有数据来源都是网络公开数据。通过对公开数据的收集，数据预处理，汇总，描述性统计等方式
熟悉相应的技术应用，一些分析的结论和更加关联的数据，有待后续的个人探索
步骤和方式：
0.了解数据限制：
01.**纳税企业或注册地在**的民办非企业、社会团体等单位
02.*** 标准： 博士每人每月2000元,硕士每人每月800元,学士和高级工每人每月600元(实际租金低于补贴标准的,按实际租金补贴),补贴期限累计不超过36个月。
1.下载数据，
可以下载到公开的公示信息：在公示的数据，注意数据完整性，数据有首批数据，第二批审核通过数据，季度补发数据等数据
2.数据预处理
1.数据格式处理
01.正常的数据格式
201*年的数据要处理成以下格式
201*年数据-添加字段-年份，以便和其他的年份做为区别，
发放区 姓名 单位名称 4月 5月 6月 补贴总金额
02.补发的数据格式: 发放区 姓名 单位名称 补发总金额（元）
2.数据变换处理
发放月份用的是对号，把对号替换成1，其没有对号的替换为0，以便汇总发放的月份
同金额相相处理后，可以判断用户的学历信息(注意实际租金低于补贴的情况，可以查看总金额的类型有哪些)
注意实际租金低于补贴的情况-以租金所在的区间为该用户所在的范围设置学历
3.数据
缺失数据处理
重复数据处理-
数据一致性处理-
201*年数据： 201*机构设置发生重大调整 2020年:
数据规范化
公司名称 --有的带有点号，有的不带
*****事务所有限公司.
3.数据汇总处理
201*年花销 **元 ，即**亿元 共计： **位补贴人员
doctor ** master ** undergraduate ** 补贴比例： 1:*:**
其中人力资源服务的公司，这部分人员的有待归类
各个地区的企业数，
申请补贴的用户数，
不同用户的申请金额等级，
各个区的不同申请等级的用户数
4.数据解释说明
各个地区的企业数，申请补贴的用户数，不同用户的申请金额等级，各个区的不同申请等级的用户数可以粗略的看到一些信息
数据进行独立检查后，要做相互验证
独立优秀的数据提取和数据清洗能力，确保数据的统计口径，以及统计口径下的数据准确性
5.不足和后续展望
1.(在编制的情况不予考虑)，所以本数据只是反映
2.本地有房的不在名单之中，本地就业无房的数据部分的比例数据没有，所以这部分数据不全
3.不是所有的人都有申请，或者申请的都有审批，这部分用户数据不全，**的符合条件的毕业生，申请的毕业生的人数不全
4.有部分年月的数据缺失，有待补充
5.企业名称，可以结合所在的位置，行业类型做进一步的数据分析和展示，通过地图的形式，可以更好的直观了解目前的企业分布
5.注意： 真实数据和理论数据有所区别，以及一旦分析数据作为考核数据会对数据参与者造成相应的引导和影响，
所以，本数据只是数据分析的一个流程展示
代码附录
#!/usr/bin/env python
# -*-coding:utf-8-*-
import pandas as pd
import numpy as np
if __name__ == '__main__':
# 数据结构示例
""" pay_area,applicant_nm,applicant_company_nm,M1月,M2月,M3月,金额合计
开发区,***,***有限公司,1,1,1,990
**区,***,**有限公司,1,0,0,800"""
"""1.读取数据，构造数据名称和索引 """
insert_file = r"F:\Database\Data_201*.txt"
insert_df = pd.read_csv(insert_file, sep=',', header=None, skiprows=1, encoding='utf-8')
insert_df.columns = ['pay_area', 'applicant_nm', 'applicant_company_nm',
'pay_m1_flag', 'pay_m2_flag', 'pay_m3_flag', 'pay_amt']
"""2.查看数据 """
print(insert_df.head())
print(insert_df.dtypes)
print(insert_df.shape)
"""3.处理数据 """
"""3.处理数据_去重- SQL- distinct """
insert_df = insert_df.drop_duplicates()
"""3.处理数据_增加列- SQL- distinct """
# insert_df['mean_flag'] = insert_df['pay_amt']/(insert_df['pay_m2_flag']+insert_df['pay_m3_flag'])
insert_df = insert_df.assign(mean_flag=insert_df['pay_amt']/(insert_df['pay_m1_flag'] + insert_df['pay_m2_flag']
+ insert_df['pay_m3_flag']))
"""3.处理数据_条件处理- SQL- case when """
# 默认左闭右闭区间-inclusive=True case when
insert_df['educational'] = np.select([insert_df['mean_flag'].between(0, 600, inclusive=True),
insert_df['mean_flag'].between(600, 800.01, inclusive=False),
insert_df['mean_flag'].between(800.01, 2500, inclusive=True)],
['undergraduate', 'master', 'doctor'],
default='Unknown')
"""3.处理数据_筛选- SQL- where """
expt = insert_df[insert_df['educational'] == 'Unknown']
"""3.处理数据_分组 SQL- group """
# group by count 在pandas中 分组操作groupby可以选择对行或者列进行拆分
# DataFrameGroupBy
# edu_sum = insert_df.groupby(['pay_area','educational']).count()
edu_sum_DataFrameGroupBy = insert_df[insert_df['educational'] == 'undergraduate']\
.groupby(['pay_area', 'educational'], axis=0, as_index=False, sort=True)['applicant_nm']
# 查看分组的个数
print('分组个数是: ', len(edu_sum_DataFrameGroupBy))
# Only relevant for DataFrame input. as_index=False is effectively “SQL-style” grouped output
edu_sum = edu_sum_DataFrameGroupBy.count()
"""3.处理数据_排序 SQL- order by """
# sort_index
# sort_values()是按值排序的方法。它接受一个by参数，它将使用要与其排序值的DataFrame的列名称。
edu_sum_df = edu_sum.sort_values(by=['applicant_nm'], ascending=False)
"""3.处理数据 """
comp_df = insert_df[['pay_area', 'applicant_company_nm']] \
.groupby(['pay_area'], axis=0, as_index=False, sort=True)['applicant_company_nm'] \
.count()
comp_df = insert_df[['pay_area', 'applicant_company_nm']]\
.drop_duplicates(subset=None, keep='first') \
.groupby(['pay_area'], axis=0, as_index=False, sort=True)['applicant_company_nm'] \
.count()
calcu_df = insert_df[['pay_area', 'applicant_company_nm', 'educational']]
# ['undergraduate', 'master', 'doctor']
comp_df = calcu_df[calcu_df['educational'] == 'undergraduate'] \
.groupby(['applicant_company_nm'], axis=0, as_index=False, sort=True)['pay_area'] \
.count()
# 企业人数分布
comp_df = comp_df.sort_values(by=['pay_area'], ascending=False).reset_index(drop=True)
print(comp_df.head(50))
# 字符串的操作
surname = pd.DataFrame(insert_df['applicant_nm'].str.strip().str[0])
# 索引 ——> 列,便于分组统计频次使用 增加列的另外一种方式
surname['index_df'] = surname.index
surname_rs = surname.groupby(['applicant_nm'], axis=0, as_index=False)['index_df'] \
.count()\
.sort_values(by=['index_df'], ascending=False).reset_index(drop=True)
print(surname_rs.head(30))
print(len(surname_rs))
"""4.数据输出 """
mid_out_path = r"F:\Database\edu_sum_df.txt"
comp_df.to_csv(mid_out_path, sep='\t', index=False, encoding='utf-8')
参考：
政府信息公开制度_人力资源和社会保障局
Pandas-Pandas中的Case when和default https://www.cnpython.com/qa/68227
pandas documentation https://pandas.pydata.org/pandas-docs/stable/reference/index.html标签：insert,index,Python,pay,applicant,df,数据,预处理,ETL 来源： https://www.cnblogs.com/ytwang/p/13740904.html