使用 pyspark 读取 csv 文件时编码错误

Question

对于我在大学的课程，我运行 pyspark-notebook docker image

docker pull jupyter/pyspark-notebook
docker run -it --rm -p 8888:8888 -v /path/to/my/working/directory:/home/jovyan/work jupyter/pyspark-notebook

然后运行下一个python代码

import pyspark 
from pyspark.sql import SparkSession
from pyspark.sql.types import *

sc = pyspark.SparkContext('local[*]')
spark = SparkSession(sc)
spark

listings_df = spark.read.csv("listings.csv", header=True, mode='DROPMALFORMED') 
# adding encoding="utf8" to the line above doesn't help also
listings_df.printSchema()

读取文件时出现问题。似乎 spark 错误地读取了我的文件（可能是因为编码问题？）并且读取后 listings_df 有 16494 行，而正确的行数是 16478（用 pandas.read_csv() 检查）。你可以看到运行ning

肯定也有东西坏了

listings_df.groupBy("room_type").count().show()

给出下一个输出

+---------------+-----+
|      room_type|count|
+---------------+-----+
|            169|    1|
|        4.88612|    1|
|        4.90075|    1|
|    Shared room|   44|
|             35|    1|
|            187|    1|
|           null|   16|
|             70|    1|
|             27|    1|
|             75|    1|
|     Hotel room|  109|
|            198|    1|
|             60|    1|
|            280|    1|
|Entire home/apt|12818|
|            220|    1|
|            190|    1|
|            156|    1|
|            450|    1|
|        4.88865|    1|
+---------------+-----+
only showing top 20 rows

而实际 room_type 值仅为 ['Private room'、'Entire home/apt'、'Hotel room'、'Shared room']。

可能有用的 Spark 信息：

SparkSession - in-memory

SparkContext

Spark UI

Version
v3.1.2
Master
local[*]
AppName
pyspark-shell

和文件的编码

!file listings.csv
listings.csv: UTF-8 Unicode text

listings.csv 是从 here

下载的 Airbnb 统计数据 csv 文件

所有运行和驱动代码我也上传到 Colab

Answer 1

我认为从这里编码文件应该可以解决问题。因此，您将 encoding="utf8" 添加到变量 listings_df.

的元组中

如下图；

listings_df = spark.read.csv("listings.csv", encoding="utf8", header=True, mode='DROPMALFORMED')

Answer 2

我发现了两件事：

有些行需要转义引号 (escape='"')
另外 @JosefZ 提到了不需要的换行符 (multiLine=True)

这就是您必须阅读的方式：

input_df = spark.read.csv(path, header=True, multiLine=True, escape='"')

output_df = input_df.groupBy("room_type").count()
output_df.show()
+---------------+-----+
|      room_type|count|
+---------------+-----+
|    Shared room|   44|
|     Hotel room|  110|
|Entire home/apt|12829|
|   Private room| 3495|
+---------------+-----+

使用 pyspark 读取 csv 文件时编码错误

Wrong encoding when reading csv file with pyspark

csv

encoding

pyspark