Besoin d'aide sur PySpark [Jupiter - Anaconda]

**Wabbina** · 29/09/2018, 21h02

Bonjour à tous,
je suis novice concernant l'utilisation de PySpark et je rencontre d'énorme difficulté à resoudre mon bug. Alors, j'ai écris un code permettant de charger un fichier texte et de renvoyer un dataframe contenant 3 colonnes :
- le mot
- sa longueur
- sa fréquence

ci-joint le code :

Code Python :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
from pyspark.context import SparkContext
from pyspark.sql import SparkSession
from pyspark.sql import Row
 
 
#création de contexte SPark
sc = SparkContext.getOrCreate()
 
#création de la session Spark
spark = SparkSession.builder.getOrCreate()
 
#filtrage des noms contenants des mots clés
def filterWord(word):
    return ("@" in word) or ("/" not in word) == True \
            and len(word) > 0
 
 
#fonction de chargement de données
def loadData(path):
    rdd = sc.textFile(path)\
            .flatMap(lambda lines : lines.lower().split())\
            .flatMap(lambda word : word.split(",")) \
            .flatMap(lambda word : word.split(".")) \
            .flatMap(lambda word : word.split(";")) \
            .flatMap(lambda word : word.split(":")) \
            .flatMap(lambda word : word.split("?")) \
            .flatMap(lambda word : word.split("!")) \
            .flatMap(lambda word : word.split("\"")) \
            .flatMap(lambda word : word.split("-")) \
            .flatMap(lambda word : word.split("\'"))\
            .filter(filterWord)\
            .map(lambda word : (word, 1)) \
            .reduceByKey( lambda count1, count2 : count1 + count2 ) \
            .map(lambda word, count : Row(label=word, length=len(word), freq=count))
 
    return spark.createDataFrame(rdd)
 
 
#chargement des données
dataFrame = loadData("C:\\Spark\\test\\iliad.txt")
print(dataFrame)
print("le nombre total de mots est :", dataFrame.count())

En exécutant le code, voici les erreurs que j'obtiens:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
<ipython-input-33-2cd728bde611> in <module>()
     23 
     24 #chargement des données
---> 25 dataFrame = loadData("C:\\Spark\\test\\iliad.txt")
     26 print(dataFrame)
     27 print("le nombre total de mots est :", dataFrame.count())

<ipython-input-33-2cd728bde611> in loadData(path)
     19     rdd = sc.textFile(path)            .flatMap(lambda lines : lines.lower().split())            .flatMap(lambda word : word.split(","))             .flatMap(lambda word : word.split("."))             .flatMap(lambda word : word.split(";"))             .flatMap(lambda word : word.split(":"))             .flatMap(lambda word : word.split("?"))             .flatMap(lambda word : word.split("!"))             .flatMap(lambda word : word.split("\""))             .flatMap(lambda word : word.split("-"))             .flatMap(lambda word : word.split("\'"))            .filter(filterWord)            .map(lambda word : (word, 1))             .reduceByKey( lambda count1, count2 : count1 + count2 )             .map(lambda word, count : Row(label=word, length=len(word), freq=count))
     20 
---> 21     return spark.createDataFrame(rdd)
     22 
     23 

~\Anaconda3\lib\site-packages\pyspark\sql\session.py in createDataFrame(self, data, schema, samplingRatio, verifySchema)
    687 
    688         if isinstance(data, RDD):
--> 689             rdd, schema = self._createFromRDD(data.map(prepare), schema, samplingRatio)
    690         else:
    691             rdd, schema = self._createFromLocal(map(prepare, data), schema)

~\Anaconda3\lib\site-packages\pyspark\sql\session.py in _createFromRDD(self, rdd, schema, samplingRatio)
    382         """
    383         if schema is None or isinstance(schema, (list, tuple)):
--> 384             struct = self._inferSchema(rdd, samplingRatio, names=schema)
    385             converter = _create_converter(struct)
    386             rdd = rdd.map(converter)

~\Anaconda3\lib\site-packages\pyspark\sql\session.py in _inferSchema(self, rdd, samplingRatio, names)
    353         :return: :class:`pyspark.sql.types.StructType`
    354         """
--> 355         first = rdd.first()
    356         if not first:
    357             raise ValueError("The first row in RDD is empty, "

~\Anaconda3\lib\site-packages\pyspark\rdd.py in first(self)
   1391         ValueError: RDD is empty
   1392         """
-> 1393         rs = self.take(1)
   1394         if rs:
   1395             return rs[0]

~\Anaconda3\lib\site-packages\pyspark\rdd.py in take(self, num)
   1373 
   1374             p = range(partsScanned, min(partsScanned + numPartsToTry, totalParts))
-> 1375             res = self.context.runJob(self, takeUpToNumLeft, p)
   1376 
   1377             items += res

~\Anaconda3\lib\site-packages\pyspark\context.py in runJob(self, rdd, partitionFunc, partitions, allowLocal)
   1011         # SparkContext#runJob.
   1012         mappedRDD = rdd.mapPartitions(partitionFunc)
-> 1013         sock_info = self._jvm.PythonRDD.runJob(self._jsc.sc(), mappedRDD._jrdd, partitions)
   1014         return list(_load_from_socket(sock_info, mappedRDD._jrdd_deserializer))
   1015 

~\Anaconda3\lib\site-packages\py4j\java_gateway.py in __call__(self, *args)
   1255         answer = self.gateway_client.send_command(command)
   1256         return_value = get_return_value(
-> 1257             answer, self.gateway_client, self.target_id, self.name)
   1258 
   1259         for temp_arg in temp_args:

~\Anaconda3\lib\site-packages\pyspark\sql\utils.py in deco(*a, **kw)
     61     def deco(*a, **kw):
     62         try:
---> 63             return f(*a, **kw)
     64         except py4j.protocol.Py4JJavaError as e:
     65             s = e.java_exception.toString()

~\Anaconda3\lib\site-packages\py4j\protocol.py in get_return_value(answer, gateway_client, target_id, name)
    326                 raise Py4JJavaError(
    327                     "An error occurred while calling {0}{1}{2}.\n".
--> 328                     format(target_id, ".", name), value)
    329             else:
    330                 raise Py4JError(

Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.runJob.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 72.0 failed 1 times, most recent failure: Lost task 0.0 in stage 72.0 (TID 102, localhost, executor driver): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "C:\Spark\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 230, in main
  File "C:\Spark\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 225, in process
  File "C:\Spark\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\serializers.py", line 372, in dump_stream
    vs = list(itertools.islice(iterator, batch))
  File "C:\Users\Tiemounou\Anaconda3\lib\site-packages\pyspark\rdd.py", line 1371, in takeUpToNumLeft
    yield next(iterator)
  File "C:\Spark\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\util.py", line 55, in wrapper
    return f(*args, **kwargs)
TypeError: <lambda>() missing 1 required positional argument: 'count'

	at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:298)
	at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:438)
	at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:421)
	at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:252)
	at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
	at scala.collection.Iterator$class.foreach(Iterator.scala:893)
	at org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28)
	at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
	at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
	at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
	at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
	at org.apache.spark.InterruptibleIterator.to(InterruptibleIterator.scala:28)
	at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
	at org.apache.spark.InterruptibleIterator.toBuffer(InterruptibleIterator.scala:28)
	at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
	at org.apache.spark.InterruptibleIterator.toArray(InterruptibleIterator.scala:28)
	at org.apache.spark.api.python.PythonRDD$$anonfun$3.apply(PythonRDD.scala:149)
	at org.apache.spark.api.python.PythonRDD$$anonfun$3.apply(PythonRDD.scala:149)
	at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
	at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
	at org.apache.spark.scheduler.Task.run(Task.scala:109)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

Driver stacktrace:
	at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1602)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1590)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1589)
	at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
	at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1589)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:831)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:831)
	at scala.Option.foreach(Option.scala:257)
	at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:831)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1823)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1772)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1761)
	at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
	at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:642)
	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2034)
	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2055)
	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2074)
	at org.apache.spark.api.python.PythonRDD$.runJob(PythonRDD.scala:149)
	at org.apache.spark.api.python.PythonRDD.runJob(PythonRDD.scala)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
	at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
	at py4j.Gateway.invoke(Gateway.java:282)
	at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
	at py4j.commands.CallCommand.execute(CallCommand.java:79)
	at py4j.GatewayConnection.run(GatewayConnection.java:238)
	at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "C:\Spark\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 230, in main
  File "C:\Spark\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 225, in process
  File "C:\Spark\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\serializers.py", line 372, in dump_stream
    vs = list(itertools.islice(iterator, batch))
  File "C:\Users\Tiemounou\Anaconda3\lib\site-packages\pyspark\rdd.py", line 1371, in takeUpToNumLeft
    yield next(iterator)
  File "C:\Spark\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\util.py", line 55, in wrapper
    return f(*args, **kwargs)
TypeError: <lambda>() missing 1 required positional argument: 'count'

	at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:298)
	at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:438)
	at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:421)
	at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:252)
	at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
	at scala.collection.Iterator$class.foreach(Iterator.scala:893)
	at org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28)
	at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
	at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
	at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
	at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
	at org.apache.spark.InterruptibleIterator.to(InterruptibleIterator.scala:28)
	at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
	at org.apache.spark.InterruptibleIterator.toBuffer(InterruptibleIterator.scala:28)
	at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
	at org.apache.spark.InterruptibleIterator.toArray(InterruptibleIterator.scala:28)
	at org.apache.spark.api.python.PythonRDD$$anonfun$3.apply(PythonRDD.scala:149)
	at org.apache.spark.api.python.PythonRDD$$anonfun$3.apply(PythonRDD.scala:149)
	at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
	at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
	at org.apache.spark.scheduler.Task.run(Task.scala:109)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	... 1 more



In [ ]:

Merci par avance de votre aide

bien cordialement,
Wab

**Wabbina** · 30/09/2018, 01h15

Problème résolu.

En fait, le problème vient de ma fonction de chargement (cf ligne 17)

Code Python :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
#fonction de chargement de données
def loadData(path):
    rdd = sc.textFile(path)\
            .flatMap(lambda lines : lines.lower().split())\
            .flatMap(lambda word : word.split(",")) \
            .flatMap(lambda word : word.split(".")) \
            .flatMap(lambda word : word.split(";")) \
            .flatMap(lambda word : word.split(":")) \
            .flatMap(lambda word : word.split("?")) \
            .flatMap(lambda word : word.split("!")) \
            .flatMap(lambda word : word.split("\"")) \
            .flatMap(lambda word : word.split("-")) \
            .flatMap(lambda word : word.split("\'"))\
            .filter(filterWord)\
            .map(lambda word : (word, 1)) \
            .reduceByKey( lambda count1, count2 : count1 + count2 ) \
            .map(lambda word, count : Row(label=word, length=len(word), freq=count))
 
    return spark.createDataFrame(rdd)

Dans le map, j'ai considéré qu'il prend en entrée 2 variables : le mot et sa fréquence. Et c'est de là d'ou vient le problème puiqu'en sortie de la fonction "reduceByKey", on a une liste de tuple (couple de variable). Du coup la solution est :

Code Python :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
#fonction de chargement de données
def loadData(path):
    rdd = sc.textFile(path)\
            .flatMap(lambda lines : lines.lower().split())\
            .flatMap(lambda word : word.split(",")) \
            .flatMap(lambda word : word.split(".")) \
            .flatMap(lambda word : word.split(";")) \
            .flatMap(lambda word : word.split(":")) \
            .flatMap(lambda word : word.split("?")) \
            .flatMap(lambda word : word.split("!")) \
            .flatMap(lambda word : word.split("\"")) \
            .flatMap(lambda word : word.split("-")) \
            .flatMap(lambda word : word.split("\'"))\
            .filter(filterWord)\
            .map(lambda word : (word, 1)) \
            .reduceByKey( lambda count1, count2 : count1 + count2 ) \
            .map(lambda data : Row(label=data[0], length=len(data[0]), freq=data[1]))
 
    return spark.createDataFrame(rdd)

voilà !!!!

**Jao2022** · 16/07/2022, 13h44

Bonjour,

J'ai un petit soucis avec mon code qui ne renvoie pas d'erreur.

La conversion ne se fait pas : la date reste au même format "annéemois" = "202012"

Je souhaite que le format devienne "01/12/2020"

Voici le code:

Code Python :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
from pyspark.sql.functions import split
# splitting the date column into day using split function
combined = combined.withColumn('month', split(df1['Period Desc'], ' ').getItem(0)) 
# display the data
combined.show(5)

Code Python :

Sélectionner tout - Visualiser dans une fenêtre à part

from pyspark.sql.functions import concat, col, lit

Code Python :

Sélectionner tout - Visualiser dans une fenêtre à part

combined=combined.withColumn("day", lit('01'))

Code Python :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
combined = combined.withColumn("month", when(combined["month"] == 'January', '01').otherwise(combined["month"]))
combined = combined.withColumn("month", when(combined["month"] == 'February', 'O2').otherwise(combined["month"]))
combined = combined.withColumn("month", when(combined["month"] == 'March', '03').otherwise(combined["month"]))
combined = combined.withColumn("month", when(combined["month"] == 'April', 'O4').otherwise(combined["month"]))
combined = combined.withColumn("month", when(combined["month"] == 'May', '05').otherwise(combined["month"]))
combined = combined.withColumn("month", when(combined["month"] == 'June', 'O6').otherwise(combined["month"]))
combined = combined.withColumn("month", when(combined["month"] == 'July', '07').otherwise(combined["month"]))
combined = combined.withColumn("month", when(combined["month"] == 'August', 'O8').otherwise(combined["month"]))
combined = combined.withColumn("month", when(combined["month"] == 'September', '09').otherwise(combined["month"]))
combined = combined.withColumn("month", when(combined["month"] == 'October', '10').otherwise(combined["month"]))
combined = combined.withColumn("month", when(combined["month"] == 'November', '11').otherwise(combined["month"]))
combined = combined.withColumn("month", when(combined["month"] == 'December', '12').otherwise(combined["month"]))

Code Python :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
combined = combined.withColumn("my", concat(col("day"), lit("/"), col("month")))
combined = combined.withColumn("final_date", concat(col("my"), lit("/"), col("Year")))
combined=combined.drop('my')
combined=combined.drop('day')
combined=combined.drop('month')

Je vous remercie pour vos précieux conseils et support.

Besoin d'aide sur PySpark [Jupiter - Anaconda]

Hadoop & co

Discussions similaires

Partager

Partager