Искра неправильно читает CSV

Question

Я хочу прочитать train.csv в спарк, но похоже, что спарк как-то неправильно читает файл.Я прочитал CSV в панды с Python, и он показывает правильное значение 1 в качестве первой записи в project_is_approved.Когда я читаю csv с помощью spark (scala), я получаю строку, предположительно, из другого места в наборе данных.

Почему это происходит?В большинстве примеров используется синтаксис, который я использую для чтения 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

Вам необходимо определить escape-символ, чтобы при синтаксическом анализе можно было игнорировать запятую (,) внутри текста

. Это можно сделать как

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.Таким образом, вместо чтения CSV-файла с помощью Python и записи его на диск, чтобы вы могли прочитать его позже, используя spark, вы можете использовать pandas перед преобразованием его в Spark Dataframe (который вы получите благодаря распределенным вычислениям на spark с использованием dataframe)

Я не очень знаком со 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, чтобы вы могли избежать потери ресурсов вычислений, и spark прочитает файл в правильном формате