MapFunction is very slow at Ray engine

[dchigarev]

System information

• OS Platform and Distribution (e.g., Linux Ubuntu 16.04): All, (Ray)
• Modin version (modin.__version__): Current master
• Python version: 3.7.5
• Code we can use to reproduce:

import numpy as np
import string
import csv
import os
from timeit import default_timer as timer

RAND_LOW = -100
RAND_HIGH = 100
N = 30000
M = 1000
TEST_FILENAME = os.path.abspath(f"int_dataset-{N},{M},{RAND_LOW},{RAND_HIGH}.csv")


def generate_data_file(filename, row_n, col_n):
    letters = list(string.ascii_letters)
    columns = [f"col{i}" for i in np.arange(col_n)]

    with open(filename, "w") as file:
        writer = csv.writer(file, delimiter=",")
        writer.writerow(columns)
        for _ in np.arange(row_n):
            writer.writerow(np.random.randint(RAND_LOW, RAND_HIGH, col_n))


if __name__ == "__main__":
    import pandas
    import modin.pandas as pd

    if not os.path.exists(TEST_FILENAME):
        generate_data_file(TEST_FILENAME, N, M)

    md_df = pd.read_csv(TEST_FILENAME)
    print("file readed")
    t1_md = timer()
    md_res = print(md_df.abs())
    t2_md = timer()

    print("Modin:", t2_md - t1_md)

Describe the problem

Related code will measure the working time of abs, which is implemented via MapFunction, that call map_partitions inside partition_manager. The problem is that, it works really slow at ray execution engine, but after changing map_partitions to lazy_map_partitions it starts to work fine:

112 cores:
	map_partitions: 2.6236
	lazy_map_partitions: 0.1817

I've run this code at python execution engine and didn't notice such a problem.
So, is just applying function to partitions is not async at Ray? Are there any limitations of using lazy_map_partitions instead of map_partitions? Can we just change map_partitions to lazy_map_partitions everywhere in modin?
Actually, the difference is not so critical for machines which have not many cores, because it has fewer partitions and so less places to possible blocks

8 cores:
	map_partitions: 0.2313
	lazy_map_partitions: 0.1215

@devin-petersohn please take a look

[devin-petersohn]

We can certainly test this for all operations. The lazy_map_partitions does make some assumptions, but it would be a good optimization for the interactive case.

MapReduceFunction might give some trouble, but I think we can work around that. Let's do this for the next release.