Spark读取合并后Pandas DataFrame 报错找不到org.apache.spark.csv怎么办

不需要单独下载org.apache.spark.csv到本地。你遇到的报错是多个问题共同导致的，包括Spark API用法错误、参数配置冗余以及代码笔误，具体问题和解决方案如下：

存在的问题 #

• 语法逻辑错误：spark.read.load() 方法接收的入参是文件存储路径，不是已经加载到内存的Pandas DataFrame对象，你直接传入合并后的df属于用法错误。
• 格式参数冗余：Spark 2.0及以上版本已经内置CSV数据源支持，不需要写全限定名org.apache.spark.csv，直接指定format("csv")即可。
• 代码笔误：你前面读取的三个DataFrame命名为dfA、dfB、dfC，删除索引列时调用的是df_7news、df_theAge等未定义变量，运行时会先触发变量不存在报错。
• 过滤逻辑列名不匹配：你定义的Excel字段包含Name、Prod_No、Category、URL、Description，没有名为Article的列，后续过滤df.Article.str.contains会触发列不存在报错。

解决方案 #

方法一：直接转换Pandas DataFrame为Spark DataFrame（最优）

不需要走CSV读取流程，直接调用Spark自带的转换方法即可：

import pandas as pd

# 读取Excel
dfA = pd.read_excel("/content/gdrive/MyDrive/Colab Notebooks/dfA.xlsx",  names=['','Name', 'Prod_No','Category','URL','Description']) 
dfB = pd.read_excel("/content/gdrive/MyDrive/Colab Notebooks/dfB.xlsx", names=['','Name', 'Prod_No','Category','URL','Description']) 
dfC = pd.read_excel("/content/gdrive/MyDrive/Colab Notebooks/dfC.xlsx", names=['','Name', 'Prod_No','Category','URL','Description']) 

# 删除冗余索引列，修正变量名
dfA = dfA.drop([''], axis=1)
dfB = dfB.drop([''], axis=1)
dfC = dfC.drop([''], axis=1)

# 合并+过滤包含Replay的行，修正列名
df = pd.concat([dfA,dfB,dfC],ignore_index=True)
df = df[~df['Description'].str.contains("Replay", na=False)]

# 直接转换为Spark DataFrame，无需读取CSV
spark_df = spark.createDataFrame(df)

方法二：先存为CSV再用Spark读取

如果必须走CSV读取流程，先把合并后的Pandas DataFrame存为本地文件，再传入文件路径读取：

# 存为本地CSV文件
df.to_csv("/tmp/merged_data.csv", index=False, encoding="utf-8")

# 用Spark读取CSV，不需要指定全限定名的数据源
spark_df = spark.read.format("csv")\
    .option("header", "true")\
    .option("mode", "PERMISSIVE")\
    .option("inferSchema", "True")\
    .load("/tmp/merged_data.csv")

如果是Spark 1.x的旧版本，需要用到第三方spark-csv包，启动pyspark时添加参数--packages com.databricks:spark-csv_2.11:1.5.0即可自动拉取依赖，不需要手动下载到本地。

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