如何用Pandas读取Excel并通过PySpark将数据插入Hive？

看起来你已经完成了Excel数据的清洗工作，接下来只需要把Pandas DataFrame转换成PySpark DataFrame，然后利用PySpark的Hive集成能力写入Hive即可。下面是具体的实现步骤和修改后的代码：

核心步骤说明 #

• 转换数据格式：PySpark无法直接操作Pandas DataFrame，需要先把清洗好的data转换成Spark DataFrame。
• 写入Hive表：使用PySpark的saveAsTable或者write API将数据写入Hive，支持创建新表、追加数据、覆盖数据等模式。

修改后的完整代码 #

版本1：基于你的现有代码（兼容Spark 1.x） #

from pyspark import SparkConf, SparkContext
from pyspark.sql import HiveContext
import pandas as pd

sparkConf = SparkConf().setAppName("ExcelToHive")
sc = SparkContext(conf=sparkConf)
sqlContext = HiveContext(sc)

excel_file = pd.ExcelFile("export_n_moreExportData10846.xls")
for sheet_name in excel_file.sheet_names:
    try:
        df = pd.read_excel(excel_file, header=None, squeeze=True, sheet_name=sheet_name)
        for i, row in df.iterrows():
            if row.notnull().all():
                data = df.iloc[(i+1):].reset_index(drop=True)
                data.columns = list(df.iloc[i])
                break
        for c in data.columns:
            data[c] = pd.to_numeric(data[c], errors='ignore')
        
        # -------------------------- 新增的写入Hive代码 --------------------------
        # 1. 将Pandas DataFrame转换为Spark DataFrame
        spark_df = sqlContext.createDataFrame(data)
        
        # 2. 定义Hive表名（这里用sheet名作为表名，你可以自定义，比如加前缀）
        hive_table_name = f"your_database.{sheet_name}"  # 替换成你的数据库名，或者直接用sheet_name
        
        # 3. 写入Hive表：这里用"overwrite"模式，也可以用"append"或者"ignore"
        spark_df.write.mode("overwrite").saveAsTable(hive_table_name)
        
        print(f"成功将sheet {sheet_name}的数据写入Hive表 {hive_table_name}")
        # ----------------------------------------------------------------------
        
    except Exception as e:
        print(f"处理sheet {sheet_name}时出错: {str(e)}")
        continue

版本2：使用SparkSession（推荐Spark 2.0+） #

from pyspark.sql import SparkSession
import pandas as pd

# 初始化SparkSession并启用Hive支持
spark = SparkSession.builder \
    .appName("ExcelToHive") \
    .enableHiveSupport() \
    .getOrCreate()

excel_file = pd.ExcelFile("export_n_moreExportData10846.xls")
for sheet_name in excel_file.sheet_names:
    try:
        df = pd.read_excel(excel_file, header=None, squeeze=True, sheet_name=sheet_name)
        for i, row in df.iterrows():
            if row.notnull().all():
                data = df.iloc[(i+1):].reset_index(drop=True)
                data.columns = list(df.iloc[i])
                break
        for c in data.columns:
            data[c] = pd.to_numeric(data[c], errors='ignore')
        
        # 转换为Spark DataFrame并写入Hive
        spark_df = spark.createDataFrame(data)
        hive_table_name = f"your_database.{sheet_name}"
        
        # 写入模式说明：
        # - overwrite: 覆盖现有表
        # - append: 追加到现有表
        # - ignore: 如果表存在则跳过
        # - error: 如果表存在则报错（默认）
        spark_df.write.mode("overwrite").saveAsTable(hive_table_name)
        
        print(f"成功将sheet {sheet_name}的数据写入Hive表 {hive_table_name}")
        
    except Exception as e:
        print(f"处理sheet {sheet_name}时出错: {str(e)}")
        continue

# 关闭SparkSession
spark.stop()

关键注意事项 #

• 表名和数据库：记得把your_database替换成你实际使用的Hive数据库名，如果不需要指定数据库，直接用sheet_name作为表名即可。
• 数据类型兼容：虽然你已经把数据转成了numeric，但转换为Spark DataFrame时，Spark会自动推断数据类型。如果有特殊类型（比如日期），可能需要手动指定Schema，避免类型推断错误。
• 写入模式：根据你的需求选择合适的写入模式，比如第一次导入用overwrite，后续增量导入用append。
• 权限问题：确保运行Spark的用户有Hive表的创建/写入权限，否则会报错。

内容的提问来源于stack exchange，提问作者Krishna