| 439 | |
| 440 | |
| 441 | def test_reduction_method(): |
| 442 | df = pd.DataFrame({"x": range(50), "y": range(50, 100)}) |
| 443 | ddf = from_pandas(df, npartitions=4) |
| 444 | |
| 445 | chunk = lambda x, val=0: (x >= val).sum() |
| 446 | agg = lambda x: x.sum() |
| 447 | |
| 448 | # Output of chunk is a scalar |
| 449 | res = ddf.x.reduction(chunk, aggregate=agg) |
| 450 | assert_eq(res, df.x.count()) |
| 451 | |
| 452 | # Output of chunk is a series |
| 453 | res = ddf.reduction(chunk, aggregate=agg) |
| 454 | assert res._name == ddf.reduction(chunk, aggregate=agg)._name |
| 455 | assert_eq(res, df.count()) |
| 456 | |
| 457 | # Test with keywords |
| 458 | res2 = ddf.reduction(chunk, aggregate=agg, chunk_kwargs={"val": 25}) |
| 459 | assert ( |
| 460 | res2._name |
| 461 | == ddf.reduction(chunk, aggregate=agg, chunk_kwargs={"val": 25})._name |
| 462 | ) |
| 463 | assert res2._name != res._name |
| 464 | assert_eq(res2, (df >= 25).sum()) |
| 465 | |
| 466 | # Output of chunk is a dataframe |
| 467 | def sum_and_count(x): |
| 468 | return pd.DataFrame({"sum": x.sum(), "count": x.count()}) |
| 469 | |
| 470 | res = ddf.reduction(sum_and_count, aggregate=lambda x: x.groupby(level=0).sum()) |
| 471 | |
| 472 | assert_eq(res, pd.DataFrame({"sum": df.sum(), "count": df.count()})) |
| 473 | |
| 474 | |
| 475 | def test_reduction_method_split_every(): |