()
| 933 | |
| 934 | |
| 935 | def test_p2p_as_delayed(): |
| 936 | pytest.importorskip("distributed") |
| 937 | pytest.importorskip("numpy") |
| 938 | pytest.importorskip("dask.dataframe") |
| 939 | from distributed.utils_test import gen_cluster |
| 940 | |
| 941 | import dask.dataframe as dd |
| 942 | |
| 943 | @gen_cluster(client=True) |
| 944 | async def _test(c, s, *workers): |
| 945 | test_df = dd.from_dict( |
| 946 | { |
| 947 | "partition": [0, 1, 2, 3, 0, 1, 2, 3], |
| 948 | "value": [1, 2, 3, 4, 5, 6, 7, 8], |
| 949 | }, |
| 950 | npartitions=2, |
| 951 | ) |
| 952 | part_df = test_df.shuffle("partition", force=True, shuffle_method="p2p") |
| 953 | |
| 954 | @delayed |
| 955 | def delayed_func(x): |
| 956 | return x |
| 957 | |
| 958 | delay_df = delayed_func(part_df) |
| 959 | dd.utils.assert_eq( |
| 960 | await c.gather(c.compute(delay_df)), |
| 961 | await c.gather(c.compute(part_df)), |
| 962 | ) |
| 963 | |
| 964 | _test() |
nothing calls this directly
no test coverage detected
searching dependent graphs…