PySpark读取HDFS中|分隔CSV为DataFrame遇阻求协助

嘿，我来帮你搞定这两个问题！咱们一步步来拆解：

问题1：读取CSV输出不符合预期 #

你提到CSV用|作为分隔符，但默认的spark.read.csv是用逗号（,）作为分隔符的，这就是读出来内容不对的核心原因。只要在读取时指定分隔符参数就行，另外你的文件没有表头，记得加上header=False来避免把第一行数据当成表头：

from pyspark.sql import SparkSession

# 先确保SparkSession初始化完成（如果还没创建的话）
spark = SparkSession.builder.appName("StockDataProcessing").getOrCreate()

# 读取文件，指定分隔符为|，明确无表头
stock1 = spark.read.csv(
    "/FileStore/tables/stockdata/companylist_noheader.csv",
    sep="|",
    header=False
)

# 给列命名（可选但推荐，方便后续操作）
# 假设你的CSV有4列：代码、公司名、行业板块、细分行业，就这么命名
stock1 = stock1.toDF("symbol", "company_name", "sector", "industry")

这样读出来的DataFrame就会正确拆分每一列了。

问题2：DataFrame调用map报错AttributeError #

这是个很常见的误区：DataFrame对象没有map方法，map是Spark RDD（弹性分布式数据集）的专属方法。如果需要做数据转换，优先用DataFrame的原生API（比如withColumn、select、自定义UDF），因为这些API经过Spark优化，性能更好。

举个简单的例子，把公司名转成大写：

from pyspark.sql.functions import upper

stock1 = stock1.withColumn("company_name_upper", upper(stock1.company_name))

如果你的逻辑特别复杂，必须用自定义的遍历操作，那可以先把DataFrame转成RDD，用完map再转回DataFrame：

# 转成RDD执行map操作
stock_rdd = stock1.rdd.map(lambda row: (row.symbol, row.company_name.upper()))

# 转回DataFrame并指定列名
new_stock_df = stock_rdd.toDF(["symbol", "company_name_upper"])

总结一下 #

• 读取非逗号分隔的CSV，一定要用sep参数指定正确的分隔符
• 无表头的CSV记得设置header=False，后续可以用toDF给列起名
• DataFrame没有map方法，优先用DataFrame的高级API，真要遍历就转成RDD操作

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