import pandas
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

# 读取Excel文件
excel_data = pandas.read_excel("dataExcel.xlsx")

# 将数据写入CSV文件
excel_data.to_csv("./Output/dataExcel.csv", index=False)

# 创建Spark会话
spark = SparkSession.builder.appName("ExcelDataCleaning").master('local').getOrCreate()

# 读取CSV文件为Spark DataFrame
df = spark.read.csv("./Output/dataExcel.csv", header=True, inferSchema=True)

# 创建一个新列作为行号
df = df.withColumn("row_id", F.monotonically_increasing_id())

# 过滤掉行号为1的行，即第二行
df_without_second_row = df.filter(df.row_id != 1).drop("row_id")
df = df_without_second_row.filter(df.row_id != 0).drop("row_id")

# 需要的列名
df = df.select('序号', '地址', '企业注册地址', '注册资金', '注册时间', '营业收入', '营业成本', '净利润', '其中：研发、试验检验费', '其中：技术（研究）开发费', '其中：支付科研人员的工资及福利费', '其中：出口总额',
               '主要业务活动或主要产品1', '主要外资来源国别或地区代码', '主要外资出资比例', '本公司是否为上市（挂牌）企业主体', '工业总产值（当年价格）', '实际上缴税费总额',
               '从业人员期末人数', '其中：博士', '其中：硕士', '具有大学本科学历（位）人员', '具有大学专科学历人员', '企业被认定为高新技术企业的时间', '上市挂牌时间', '其中：技术转让收入')

# 输出列名
print(df.columns)

# 删除'地址'列为'qingxiubgs2014@sina.com'或null的行
df = df.filter((F.col('地址') != 'qingxiubgs2014@sina.com') & (F.col('地址').isNotNull()))

# 填充某列空值
df = df.fillna({'本公司是否为上市（挂牌）企业主体': '0'})
df = df.fillna({'主要业务活动或主要产品1': '无'})
df = df.fillna({'主要外资来源国别或地区代码': '0'})
df = df.fillna({'上市挂牌时间': '0'})

# 去除重复项
df = df.dropDuplicates()

# 根据序号排序
df = df.withColumn('序号', F.round(F.col('序号')).cast('integer'))
df = df.orderBy('序号')

# 将Spark DataFrame转换为Pandas DataFrame
pd = df.toPandas()

# 将Pandas DataFrame保存为CSV文件
pd.to_csv('./Output/数据清洗结果.csv', index=False)