Spark 错误地读取 CSV

Question

我想在 Spark 中读取 train.csv，但似乎 Spark 以某种方式错误地读取了文件。我用python将csv读入熊猫，它显示正确的值1作为project_is_approved中的第一个条目。当我用 spark (scala( 读取 csv 时，我得到一个字符串，大概来自数据集中的其他地方。

为什么会这样？大多数示例使用我用于读取 csv 的语法。

jakeu123@azure3:~$ python
Python 2.7.12 (default, Dec  4 2017, 14:50:18) 
[GCC 5.4.0 20160609] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>> import requests
>>> f = requests.get("https://www.dropbox.com/s/2hdbltrl8bh6kbu/train.csv?raw=1", stream=True)
>>> with open("train.csv", "w") as csv:
...     csv.write(f)
... 
Traceback (most recent call last):
File "<stdin>", line 2, in <module>
TypeError: expected a string or other character buffer object
>>> with open("train.csv", "w") as csv:
...     csv.write(f.content)
... 
>>> import pandas as pd
>>> df = pd.read_csv("train.csv")
>>> df[["project_is_approved"]].head(1)
project_is_approved
0                    1
>>> 
jakeu123@azure3:~$ ./spark/bin/spark-shell
2018-06-07 23:55:02 WARN  NativeCodeLoader:62 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
2018-06-07 23:55:09 WARN  Utils:66 - Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
2018-06-07 23:55:09 WARN  Utils:66 - Service 'SparkUI' could not bind on port 4041. Attempting port 4042.
Spark context Web UI available at http://azure3:4042
Spark context available as 'sc' (master = local[*], app id = local-1528415709241).
Spark session available as 'spark'.
Welcome to
____              __
/ __/__  ___ _____/ /__
_ / _ / _ `/ __/  '_/
/___/ .__/_,_/_/ /_/_   version 2.3.0
/_/
Using Scala version 2.11.8 (OpenJDK 64-Bit Server VM, Java 1.8.0_171)
Type in expressions to have them evaluated.
Type :help for more information.
scala> val df = spark.read.option("header", true).csv("train.csv")
2018-06-07 23:55:27 WARN  ObjectStore:568 - Failed to get database global_temp, returning NoSuchObjectException
df: org.apache.spark.sql.DataFrame = [id: string, teacher_id: string ... 14 more fields]
scala> df.select($"project_is_approved").show(1)
+--------------------+                                                          
| project_is_approved|
+--------------------+
|I currently have ...|
+--------------------+
only showing top 1 row

scala> :quit

Answer 1

您需要定义转义字符，以便在解析时可以忽略文本中的逗号(，(

这可以作为

spark.read.option("escape",""")

工作示例：

scala> val df = spark.read.option("header",true).option("escape",""").csv("train.csv");
df: org.apache.spark.sql.DataFrame = [id: string, teacher_id: string ... 14 more fields]
scala> df.select($"project_is_approved").show
+-------------------+
|project_is_approved|
+-------------------+
|                  1|
|                  0|
|                  1|
|                  0|
|                  1|
|                  1|
|                  1|
|                  1|
|                  1|
|                  1|
|                  1|
|                  1|
|                  1|
|                  0|
|                  1|
|                  1|
|                  1|
|                  1|
|                  1|
|                  0|
+-------------------+
only showing top 20 rows

Answer 2

据我所知，Spark不能直接读取URL格式的文件。因此，与其使用 python 读取 CSV 文件并将其写入磁盘以便以后可以使用 Spark 读取它，不如在将其转换为 Spark 数据帧之前使用 pandas(通过使用数据帧，您将获得 Spark 分布式计算的好处(

我对Scala不是很熟悉，所以我试图使用pyspark来解决它。

import pandas as pd
from pyspark.sql.types import StringType, LongType
frame = pd.read_csv("https://www.dropbox.com/s/2hdbltrl8bh6kbu/train.csv?raw=1", index_col=None, header=0, encoding='utf-8') # Prod
frame = frame.where(pd.notnull(frame), None) # Fill NaN with Null
schema = StructType([
StructField("id", StringType(), True),
StructField("teacher_id", StringType(), True),
StructField("teacher_prefix", StringType(), True),
StructField("school_state", StringType(), True),
StructField("project_submitted_datetime", StringType(), True),
StructField("project_grade_category", StringType(), True),
StructField("project_subject_categories", StringType(), True),
StructField("project_subject_subcategories", StringType(), True),
StructField("project_title", StringType(), True),
StructField("project_essay_1", StringType(), True),
StructField("project_essay_2", StringType(), True),
StructField("project_essay_3", StringType(), True),
StructField("project_essay_4", StringType(), True),
StructField("project_resource_summary", StringType(), True),
StructField("teacher_number_of_previously_posted_projects", LongType(), True),
StructField("project_is_approved", LongType(), True),
])
df = spark.createDataFrame(frame, schema)

哦，顺便说一句，我认为在读取 CSV 文件时提供架构是必须的，因为它不会启动任何火花作业，因此您可以避免浪费计算资源，并且 spark 将以正确的格式读取文件