dask / dask/distributed

`SystemError: unknown opcode` with compiled nested function

Open
#5,331 4 comments 0 reactions 0 assignees View on GitHub
Dominant language
Python
Stars
1.7k
Forks
778
Avg merge
2h 50m
Merged PRs (30d)
3

Description

I am encountering a `SystemError: unknown opcode` when trying to use a [nuitka](https://github.com/Nuitka/Nuitka) compiled library in a Dask cluster (both local and distributed). The source code works fine when run outside of a cluster, and when compiled by nuitka. It even works if I use cloudpickle to dump and load the outer compiled function before using it, outside of a cluster context.

I originally posted this issue to the nuitka repo [here](https://github.com/Nuitka/Nuitka/issues/1209), but after further debugging it appears that it is an issue with how cloudpickle handles compiled nested functions when they are inside of the Dask task graph representation. The problematic operation in the task graph is "apply f", which it appears cloudpickle attempts to serialize _by value_.

Versions:
- nuitka 0.6.16.4
- python 3.9.4
- pandas 1.3.3
- numpy 1.21.2
- dask 2021.9.0
- distributed 2021.9.0

`dev.py`
```python3
def process_df(df):
def f(arg):
return arg

df['a'] = df['a'].apply(f, meta=('a', int))
return df
```

Compiled with the following on Ubuntu 16.04.7
`nuitka3 --no-pyi-file --module dev.py`

Example:
```python3
import pandas as pd
import dask.dataframe as dd
from distributed import Client

df = dd.from_pandas(pd.DataFrame({'a': [1]}), npartitions=1)

from dev import process_df

print('Without Cluster:')
print(process_df(df).compute())

# this will create a local process-based Dask cluster and use it for subsequent dask dataframe operations
client = Client(processes=False)
print()

print('With Cluster:')
print(process_df(df).compute())
```

Output:
> Without Cluster:
a
0 1

>With Cluster:
XXX lineno: 2, opcode: 0
distributed.worker - WARNING - Compute Failed
Function: subgraph_callable-6a3708df-626b-4826-b785-71208e5d
args: ( a
0 1)
kwargs: {}
Exception: SystemError('unknown opcode')
SystemError Traceback (most recent call last)
/tmp/ipykernel_1446/3196535699.py in
12 print()
13 print('With Cluster:')
---> 14 print(process_df(df).compute())
...
/usr/local/lib/python3.9/dist-packages/pandas/core/apply.py in apply_standard()
1097 # List[Union[Callable[..., Any], str]]]]]"; expected
1098 # "Callable[[Any], Any]"
-> 1099 mapped = lib.map_infer(
1100 values,
1101 f, # type: ignore[arg-type]
/usr/local/lib/python3.9/dist-packages/pandas/_libs/lib.pyx in pandas._libs.lib.map_infer()
2857 result[i] = arr[i]
2858 continue
-> 2859 val = f(arr[i])
2860
2861 if cnp.PyArray_IsZeroDim(val):
/exos/dev.cpython-39-x86_64-linux-gnu.so in f()
SystemError: unknown opcode

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.