I'm new with Spark. During learning, I found whether using spark-submit or change theparameter --master led to different results.
Here is my PySpark code test.py:
import heapq
from itertools import chain
from pyspark import SparkContext
from pyspark.sql import Row, SparkSession
sc = SparkContext()
data = sc.parallelize([
(key, Row(id=i, weight=i))
for key in ['key1', 'key2']
for i in range(20)
])
def append(a, b):
a.append(b)
return a
def merge(a, b):
return heapq.nlargest(10, chain(a, b), key=lambda x: x["weight"])
data = data.combineByKey(lambda x: [x], append, merge)
print(data.collect())
When I directly run it or using spark-submit --master 'local[*]' --deploy-mode client test.py, I can get the right outcome:
[('key1', [Row(id=19, weight=19), Row(id=18, weight=18), Row(id=17, weight=17), Row(id=16, weight=16), Row(id=15, weight=15), Row(id=14, weight=14), Row(id=13, weight=13), Row(id=12, weight=12), Row(id=11, weight=11), Row(id=10, weight=10)]), ('key2', [Row(id=19, weight=19), Row(id=18, weight=18), Row(id=17, weight=17), Row(id=16, weight=16), Row(id=15, weight=15), Row(id=14, weight=14), Row(id=13, weight=13), Row(id=12, weight=12), Row(id=11, weight=11), Row(id=10, weight=10)])]
However, when I use spark-submit --master 'local' --deploy-mode client test.py, there is something wrong:
[('key1', [Row(id=0, weight=0), Row(id=1, weight=1), Row(id=2, weight=2), Row(id=3, weight=3), Row(id=4, weight=4), Row(id=5, weight=5), Row(id=6, weight=6), Row(id=7, weight=7), Row(id=8, weight=8), Row(id=9, weight=9), Row(id=10, weight=10), Row(id=11, weight=11), Row(id=12, weight=12), Row(id=13, weight=13), Row(id=14, weight=14), Row(id=15, weight=15), Row(id=16, weight=16), Row(id=17, weight=17), Row(id=18, weight=18), Row(id=19, weight=19)]), ('key2', [Row(id=0, weight=0), Row(id=1, weight=1), Row(id=2, weight=2), Row(id=3, weight=3), Row(id=4, weight=4), Row(id=5, weight=5), Row(id=6, weight=6), Row(id=7, weight=7), Row(id=8, weight=8), Row(id=9, weight=9), Row(id=10, weight=10), Row(id=11, weight=11), Row(id=12, weight=12), Row(id=13, weight=13), Row(id=14, weight=14), Row(id=15, weight=15), Row(id=16, weight=16), Row(id=17, weight=17), Row(id=18, weight=18), Row(id=19, weight=19)])]
It seems that the function in the third parameter of combineByKey didn't work. Is it due to the parameter local in spark-submit?
If so, why did I also get the right outcome when directly running python test.py?
The version of my spark is 3.2.1