मानचित्र को कॉल करने के बाद Pyspark EOFError

Question

मैं & pyspark स्पार्क करने के लिए नया हूँ।

मैं डेटाफ्रेम में एक छोटी सीएसवी फ़ाइल (~ 40k) पढ़ रहा हूं।

from pyspark.sql import functions as F 
df = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('/tmp/sm.csv') 
df = df.withColumn('verified', F.when(df['verified'] == 'Y', 1).otherwise(0)) 
df2 = df.map(lambda x: Row(label=float(x[0]), features=Vectors.dense(x[1:]))).toDF() 

मैं कुछ अजीब त्रुटि है कि हर बार नहीं होती है, लेकिन बहुत नियमित रूप से होता है

>>> df2.show(1) 
+--------------------+---------+ 
|   features| label| 
+--------------------+---------+ 
|[0.0,0.0,0.0,0.0,...|4700734.0| 
+--------------------+---------+ 
only showing top 1 row 

>>> df2.count() 
41999                   
>>> df2.show(1) 
+--------------------+---------+ 
|   features| label| 
+--------------------+---------+ 
|[0.0,0.0,0.0,0.0,...|4700734.0| 
+--------------------+---------+ 
only showing top 1 row 

>>> df2.count() 
41999                   
>>> df2.show(1) 
Traceback (most recent call last): 
    File "spark-1.6.1/python/lib/pyspark.zip/pyspark/daemon.py", line 157, in manager 
    File "spark-1.6.1/python/lib/pyspark.zip/pyspark/daemon.py", line 61, in worker  
    File "spark-1.6.1/python/lib/pyspark.zip/pyspark/worker.py", line 136, in main 
    if read_int(infile) == SpecialLengths.END_OF_STREAM: 
    File "spark-1.6.1/python/lib/pyspark.zip/pyspark/serializers.py", line 545, in read_int 
    raise EOFError 
EOFError 
+--------------------+---------+ 
|   features| label| 
+--------------------+---------+ 
|[0.0,0.0,0.0,0.0,...|4700734.0| 
+--------------------+---------+ 
only showing top 1 row 

एक बार है कि EOFError उठाया गया हो, मैं इसे फिर से नहीं देख सकेंगे जब तक मैं कुछ करना है कि चिंगारी सर्वर

के साथ बातचीत की आवश्यकता है जब मैं df2.count फोन() यह पता चलता है कि [स्टेज XXX] शीघ्र है जो मैं इसे चिंगारी सर्वर पर जाकर मतलब है। ऐसा कुछ भी जो ट्रिगर करता है, अंत में जब मैं df2 के साथ कुछ करता हूं तो EOFError को फिर से समाप्त करना प्रतीत होता है।

यह डीएफ (बनाम डीएफ 2) के साथ ऐसा प्रतीत नहीं होता है ऐसा लगता है कि यह df.map() लाइन के साथ कुछ हो रहा है।

Answer 1

क्या आप डेटा फ्रेम को आरडीडी में परिवर्तित करने के बाद मानचित्र करने का प्रयास कर सकते हैं। आप एक dataframe नक्शे पर समारोह आवेदन कर रहे हैं और उसके बाद फिर बनाने की तरह

df.rdd.map().toDF() 

कृपया मुझे पता है अगर यह काम करता है that.Syntax से एक dataframe होगा। धन्यवाद।

Answer 2

मेरा मानना ​​है कि आप स्पार्क 2.x और ऊपर चल रहे हैं। नीचे दिए गए कोड सीएसवी से अपने dataframe बनाना चाहिए:

df = spark.read.format("csv").option("header", "true").load("csvfile.csv") 

तो आप नीचे दिए गए कोड हो सकता है:

df = df.withColumn('verified', F.when(df['verified'] == 'Y', 1).otherwise(0)) 

और फिर आप पंक्ति और toDF() के बिना df2 बना सकते हैं

मुझे अगर जानते हैं यह काम करता है या यदि आप स्पार्क 1.6 का उपयोग कर रहे हैं ... धन्यवाद।