<jupyter_start><jupyter_text>Getting Started with Ray Core For a book on distributed Python, it's not without a certain irony that Python on its own islargely ineffective for distributed computing.Its interpreter is effectively single threaded which makes it difficult to e.g. leverage multiple CPUs onthe same machine, let alone a whole cluster of machines, using plain Python.That means you need extra tooling, and luckily the Python ecosystem has some options for you.For instance, libraries like `multiprocessing` can help you distribute work on a single machine, but not beyond.<jupyter_code>import ray
ray.init()
# tag::retrieve[]
import time

database = [  # <1>
    "Learning", "Ray",
    "Flexible", "Distributed", "Python", "for", "Data", "Science"
]


def retrieve(item):
    time.sleep(item / 10.)  # <2>
    return item, database[item]
# end::retrieve[]
# tag::duration[]
def print_runtime(input_data, start_time, decimals=1):
    print(f'Runtime: {time.time() - start_time:.{decimals}f} seconds, data:')
    print(*input_data, sep="\n")


start = time.time()
data = [retrieve(item) for item in range(8)]  # <1>
print_runtime(data, start)  # <2>
# end::duration[]
# tag::remote[]
@ray.remote  # <1>
def retrieve_task(item):
    return retrieve(item)  # <2>
# end::remote[]
# tag::duration_remote[]
start = time.time()
data_references = [retrieve_task.remote(item) for item in range(8)]  # <1>
data = ray.get(data_references)  # <2>
print_runtime(data, start, 2)
# end::duration_remote[]
# tag::object_store[]
database_object_ref = ray.put(database)  # <1>


@ray.remote
def retrieve_task(item):
    obj_store_data = ray.get(database_object_ref)  # <2>
    time.sleep(item / 10.)
    return item, obj_store_data[item]
# end::object_store[]
# tag::duration_object_store[]
start = time.time()
data_references = [retrieve_task.remote(item) for item in range(8)]
all_data = []

while len(data_references) > 0:  # <1>
    finished, data_references = ray.wait(data_references, num_returns=2, timeout=7.0)  # <2>
    data = ray.get(finished)
    print_runtime(data, start, 3)  # <3>
    all_data.extend(data)  # <4>

# end::duration_object_store[]
print_runtime(all_data, start)
# tag::task_dependency[]
@ray.remote
def follow_up_task(retrieve_result):  # <1>
    original_item, _ = retrieve_result
    follow_up_result = retrieve(original_item + 1)  # <2>
    return retrieve_result, follow_up_result  # <3>


retrieve_refs = [retrieve_task.remote(item) for item in [0, 2, 4, 6]]
follow_up_refs = [follow_up_task.remote(ref) for ref in retrieve_refs]  # <4>

result = [print(data) for data in ray.get(follow_up_refs)]
# end::task_dependency[]
# tag::actors[]
@ray.remote  # <1>
class DataTracker:
    def __init__(self):
        self._counts = 0

    def increment(self):
        self._counts += 1

    def counts(self):
        return self._counts
# end::actors[]
# tag::actors_remote[]
@ray.remote
def retrieve_tracker_task(item, tracker):  # <1>
    obj_store_data = ray.get(database_object_ref)
    time.sleep(item / 10.)
    tracker.increment.remote()  # <2>
    return item, obj_store_data[item]


tracker = DataTracker.remote()  # <3>

data_references = [retrieve_tracker_task.remote(item, tracker) for item in range(8)]  # <4>
data = ray.get(data_references)
print(ray.get(tracker.counts.remote()))  # <5>
# end::actors_remote[]
# tag::ownership[]
@ray.remote
def task_owned():
    return


@ray.remote
def task(dependency):
    res_owned = task_owned.remote()
    return


val = ray.put("value")
res = task.remote(dependency=val)
# end::ownership[]<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13943779571388812, 'fraction_numerical': 0.013359309768995268, 'mean_word_length': 4.7504, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 30, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '194993', 'license': 'MIT', 'n_tokens_mistral': 1287, 'n_tokens_neox': 1231, 'n_words': 310, 'path': 'notebooks/ch_02_ray_core.ipynb', 'repo_name': 'yoavfreund/learning_ray', 'size': '3593'}