Vaex DataFrame按计算字段执行Groupby操作时抛出错误

问题背景 #

通过自定义函数convert_curr结合Vaex的apply方法生成Amount_INR列后，执行分组求和操作时抛出错误。

生成Amount_INR列的代码 #

def convert_curr(x,y,z):
   c = CurrencyRates()
   return c.convert(x, 'INR', y, z)

data_df_usd['Amount_INR'] = data_df_usd.apply(convert_curr,arguments=[data_df_usd.CURRENCY_CODE,data_df_usd.TOTAL_AMOUNT,data_df_usd.SUBSCRIPTION_START_DATE_DATE])

分组求和代码 #

data_df_usd.groupby('CONTENTID', agg={'Revenue':vaex.agg.sum('Amount_INR')})

报错信息 #

RemoteTraceback: 
"""
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/dist-packages/vaex/scopes.py", line 113, in evaluate
    result = self[expression]
  File "/usr/local/lib/python3.7/dist-packages/vaex/scopes.py", line 198, in __getitem__
    raise KeyError("Unknown variables or column: %r" % (variable,))

**KeyError: "Unknown variables or column: 'lambda_function(CURRENCY_CODE, TOTAL_AMOUNT, SUBSCRIPTION_START_DATE_DATE)'"**

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.7/dist-packages/forex_python/converter.py", line 103, in convert
    converted_amount = rate * amount
TypeError: can't multiply sequence by non-int of type 'float'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/lib/python3.7/multiprocessing/pool.py", line 121, in worker
    result = (True, func(*args, **kwds))
  File "/usr/local/lib/python3.7/dist-packages/vaex/expression.py", line 1616, in _apply
    scalar_result = self.f(*[fix_type(k[i]) for k in args], **{key: value[i] for key, value in kwargs.items()})
  File "<ipython-input-7-8cc933ccf57d>", line 3, in convert_curr
    return c.convert(x, 'INR', y, z)
  File "/usr/local/lib/python3.7/dist-packages/forex_python/converter.py", line 107, in convert
    "convert requires amount parameter is of type Decimal when force_decimal=True")
forex_python.converter.DecimalFloatMismatchError: convert requires amount parameter is of type Decimal when force_decimal=True
"""


The above exception was the direct cause of the following exception:

DecimalFloatMismatchError                 Traceback (most recent call last)

<ipython-input-13-cc7b243be138> in <module>
----> 1 data_df_usd.groupby('CONTENTID', agg={'Revenue':vaex.agg.sum('Amount_INR')})

错误原因 #

1. KeyError：Vaex的apply默认生成延迟计算的表达式列，不会立即计算数值。分组时Vaex尝试重新解析这个表达式，但找不到临时生成的lambda函数引用，导致报错。
2. DecimalFloatMismatchError：forex_python的convert方法默认开启force_decimal=True，要求金额参数必须是Decimal类型，但传入的TOTAL_AMOUNT是浮点型，类型不匹配触发错误。

解决步骤 #

1. 强制生成物理列而非延迟表达式 #

在apply时添加参数强制立即计算，或者显式调用evaluate()生成实际数值列：

# 方法1：添加progress=True触发立即计算
data_df_usd['Amount_INR'] = data_df_usd.apply(convert_curr, 
                                              arguments=[data_df_usd.CURRENCY_CODE, 
                                                         data_df_usd.TOTAL_AMOUNT, 
                                                         data_df_usd.SUBSCRIPTION_START_DATE_DATE],
                                              progress=True)

# 方法2：显式调用evaluate()计算
data_df_usd['Amount_INR'] = data_df_usd.apply(convert_curr, 
                                              arguments=[data_df_usd.CURRENCY_CODE, 
                                                         data_df_usd.TOTAL_AMOUNT, 
                                                         data_df_usd.SUBSCRIPTION_START_DATE_DATE]).evaluate()

2. 修复参数类型不匹配 #

修改convert_curr函数，将金额转为Decimal类型，同时确保日期格式正确：

from decimal import Decimal
from datetime import datetime
from forex_python.converter import CurrencyRates

def convert_curr(x,y,z):
   c = CurrencyRates()
   # 将浮点型金额转为Decimal
   amount_dec = Decimal(str(y))
   # 确保日期是date对象，若为字符串则解析
   date_obj = z if isinstance(z, datetime.date) else datetime.strptime(z, '%Y-%m-%d').date()
   return c.convert(x, 'INR', amount_dec, date_obj)

3. 改用Vaex矢量化操作（性能更优） #

注册矢量化函数，避免逐行apply的性能瓶颈：

import vaex
from decimal import Decimal
from datetime import datetime
from forex_python.converter import CurrencyRates

c = CurrencyRates()

@vaex.register_function()
def convert_curr_vec(currency, amount, date):
    results = []
    for curr, amt, dt in zip(currency, amount, date):
        amt_dec = Decimal(str(amt))
        dt_obj = dt if isinstance(dt, datetime.date) else datetime.strptime(dt, '%Y-%m-%d').date()
        results.append(c.convert(curr, 'INR', amt_dec, dt_obj))
    return results

# 生成列并强制计算
data_df_usd['Amount_INR'] = data_df_usd.convert_curr_vec(data_df_usd.CURRENCY_CODE, 
                                                         data_df_usd.TOTAL_AMOUNT, 
                                                         data_df_usd.SUBSCRIPTION_START_DATE_DATE).evaluate()

4. 重新执行分组操作 #

完成上述修改后，再次运行分组代码即可正常执行：

result_df = data_df_usd.groupby('CONTENTID', agg={'Revenue': vaex.agg.sum('Amount_INR')})
print(result_df)

内容的提问来源于stack exchange，提问作者RameJ