1

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

1
  • This is happening because in Python 2.7, dict.keys() returns a list but in Python 3 it returns a dict_keys object. You may be able to get around the problem by wrapping the call to dict.keys() in a list (as in list(mydict.keys())) which is what lib2to3 would 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 a dict_keys object.
    – BoarGules
    Oct 5, 2021 at 10:42

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy