Check that map_partitions can handle a delayed partition of a dataframe input
()
| 772 | |
| 773 | |
| 774 | def test_map_partitions_df_input(): |
| 775 | """ |
| 776 | Check that map_partitions can handle a delayed |
| 777 | partition of a dataframe input |
| 778 | """ |
| 779 | pd = pytest.importorskip("pandas") |
| 780 | dd = pytest.importorskip("dask.dataframe") |
| 781 | pytest.xfail("map partitions can't deal with delayed properly") |
| 782 | |
| 783 | def f(d, a): |
| 784 | assert isinstance(d, pd.DataFrame) |
| 785 | assert isinstance(a, pd.DataFrame) |
| 786 | return d |
| 787 | |
| 788 | def main(): |
| 789 | item_df = dd.from_pandas(pd.DataFrame({"a": range(10)}), npartitions=1) |
| 790 | ddf = item_df.to_delayed()[0].persist() |
| 791 | merged_df = dd.from_pandas(pd.DataFrame({"b": range(10)}), npartitions=1) |
| 792 | |
| 793 | # Notice, we include a shuffle in order to trigger a complex culling |
| 794 | merged_df = merged_df.shuffle(on="b", shuffle_method="tasks") |
| 795 | |
| 796 | merged_df.map_partitions( |
| 797 | f, ddf, meta=merged_df, enforce_metadata=False |
| 798 | ).compute() |
| 799 | |
| 800 | with distributed.LocalCluster( |
| 801 | scheduler_port=0, |
| 802 | # Explicitly disabling dashboard to prevent related warnings being |
| 803 | # elevated to errors until `bokeh=3` is fully supported. |
| 804 | # See https://github.com/dask/dask/issues/9686 and |
| 805 | # https://github.com/dask/distributed/issues/7173 for details. |
| 806 | dashboard_address=":0", |
| 807 | scheduler_kwargs={"dashboard": False}, |
| 808 | asynchronous=False, |
| 809 | n_workers=1, |
| 810 | nthreads=1, |
| 811 | processes=False, |
| 812 | ) as cluster: |
| 813 | with distributed.Client(cluster, asynchronous=False): |
| 814 | main() |
| 815 | |
| 816 | |
| 817 | @pytest.mark.filterwarnings( |