如何使用 panda_udf 对系列运行多个函数

Posted 2023-04-13

【中文标题】如何使用 panda_udf 对系列运行多个函数

【英文标题】：How to run multiple functions against series using panda_udf

【发布时间】：2019-06-25 15:07:15

【问题描述】：

我有大量从二进制编码的图像文件。我正在尝试对它们进行解码，然后将它们保存到 hdfs。我有代码可以在 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 包装代码时，我收到一个错误，它期望返回一个系列而不是一个浮点数。我知道标量 pandas udf 输出一个系列，但我不知道如何更改代码以使其工作。

【参考方案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)