I am porting some pyspark jobs that currently run on Amazon EMR to Amazon Sagemaker. But while executing one such pyspark job, I came across the following issue:
Traceback (most recent call last):
File "/opt/ml/processing/input/code/microsegment.py", line 297, in <module>
raise exc
File "/opt/ml/processing/input/code/microsegment.py", line 282, in <module>
results = calculate_microsegments(dataset_location, kpis, deli)
File "/opt/ml/processing/input/code/microsegment.py", line 100, in calculate_microsegments
eids = rdd_splitted.flatMap(lambda x: x[len(x) - 2].split("|")).distinct().collect()
File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 413, in distinct
File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 1625, in reduceByKey
File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 1865, in combineByKey
File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 1802, in partitionBy
File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2532, in _jrdd
File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2434, in _wrap_function
File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2420, in _prepare_for_python_RDD
File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 607, in dumps
_pickle.PicklingError: Could not serialize object: TypeError: can't pickle dict_keys objects
The following is the relevant code snippet:
sc = SparkContext(appName='microsegment-job')
rdd = sc.textFile(dataset_path)
header = rdd.first()
cols = header.split(delimiter)
cols_dict = {cols[i]: i for i in range(len(cols))}
other_kpis = []
for i in cols:
if i in filter_cond or i == 'CONSUMER_ID':
continue
if i == 'eventids':
break
other_kpis.append(i)
def return_row(z):
temp = z.split(delimiter)
row = [float(temp[cols_dict[kpi]]) if temp[cols_dict[kpi]] != '' else 0.0 for kpi in (kpi_list + other_kpis)] + [
temp[cols_dict['eventids']]] + [temp[cols_dict['conversions']]]
return row
rdd_splitted = rdd.filter(lambda z: header not in z).map(return_row)
eids = rdd_splitted.flatMap(lambda x: x[len(x) - 2].split("|")).distinct().collect() # Exception thrown in this line
The Sagemaker image runs on Python 3.7 and PySpark 2.4. Also attempted with a different image that runs on Python 3.7 and PySpark 3.0. Both resulting in same error.
The EMR cluster runs on Python 2.7.16 and PySpark 2.4.4
dict.keys()returns alistbut in Python 3 it returns adict_keysobject. You may be able to get around the problem by wrapping the call todict.keys()in a list (as inlist(mydict.keys())) which is whatlib2to3would do. I can't see that call in the code you present, but I don't think there is any other obvious way to produce adict_keysobject.