Как запустить несколько функций для серии, используя panda_udf

Question

У меня есть большое количество файлов изображений, которые были закодированы из двоичного файла. Я пытаюсь расшифровать их, а затем сохранить их в формате hdf. У меня есть код, который делает это успешно в Python, но из-за большого количества файлов, которые необходимо сделать, нет смысла импортировать все, преобразовать в panda, а затем запустить мой код, даже если он все поместится на один узел как панда, которого не будет. Я пытался преобразовать свой код из функции panda в pyspark panda_udf, но безуспешно. Ниже мой код. если у кого-то есть идеи, я был бы очень признателен!

def image_save(bin,file):
    start = datetime.now()
    b64_encoded_img_binary = bin
    img_binary = np.fromstring(b64_encoded_img_binary.decode('base64'), np.uint8)
    img = cv2.imdecode(img_binary, cv2.IMREAD_COLOR)
    cv2.imwrite('/dsaa/ml_shared/image_output/'+ str(file).replace(' ','').replace('JPEG','jpg'), img)
    end = datetime.now()
    time = (end - start).total_seconds()
    return time

df['seconds'] = df.apply(lambda x: 
image_save(x['b64_encoded_image_binary'],x['file_name']), axis=1)

Код работает с пандой и правильно сохраняет файлы. Но когда я пытаюсь обернуть код с помощью @pandas_udf, я получаю сообщение об ошибке, что он ожидает, что серия будет возвращаться вместо float. Я знаю, что скалярные панды udf выводят серию, но я не могу понять, как внести изменения в код, чтобы он работал.

Answer 1

Я обдумывал это. Я не думал сделать следующий логический шаг, что панда состоит из серии. Так что это сработало, я мог взять входные данные серии и сделать их кадром данных внутри функции.

def FPD(a, b):

    def F2(a, b):
        c = cv2.imdecode(np.fromstring(a.decode('base64'), np.uint8), cv2.IMREAD_COLOR)
        cv2.imwrite('/dsaa/ml_shared/image_output/' + b, c)
        return(b)

    pdf = pd.DataFrame({'a': a, 'b' : b })
    pdf.loc[:, 'c'] = pdf.apply(lambda x: F2(x['a'], x['b']), axis = 1)
    return(b)