<jupyter_start><jupyter_text>DoFn lifecycle in Apache BeamLet's explore how `DoFn`s work in Apache Beam, and what are the different steps that are followed when executing a `DoFn`.  Install Apache Beam<jupyter_code>!pip install apache_beam[interactive]==2.28.0 --quiet<jupyter_output>[K     |████████████████████████████████| 9.0MB 8.3MB/s 
[K     |████████████████████████████████| 153kB 40.7MB/s 
[K     |████████████████████████████████| 61kB 7.2MB/s 
[K     |████████████████████████████████| 2.2MB 54.5MB/s 
[K     |████████████████████████████████| 829kB 52.2MB/s 
[K     |████████████████████████████████| 17.7MB 221kB/s 
[K     |████████████████████████████████| 61kB 7.4MB/s 
[K     |████████████████████████████████| 122kB 52.0MB/s 
[K     |████████████████████████████████| 788kB 27.1MB/s 
[K     |████████████████████████████████| 122kB 46.1MB/s 
[K     |████████████████████████████████| 112kB 48.8MB/s 
[K     |████████████████████████████████| 368kB 57.3MB/s 
[?25h  Building wheel for dill (setup.py) ... [?25l[?25hdone
  Building wheel for avro-python3 (setup.py) ... [?25l[?25hdone
  Building wheel for future (setup.py) ... [?25l[?25hdone
  Building wheel for timeloop (setup.py) ... [?25l[?25hdone
[31mERROR: multiprocess 0.70.11.1 has requir[...]<jupyter_text>Some imports<jupyter_code>import apache_beam as beam
from apache_beam.runners.interactive.interactive_runner import InteractiveRunner
import apache_beam.runners.interactive.interactive_beam as ib

from datetime import datetime<jupyter_output><empty_output><jupyter_text>The friends of `ParDo``For many situations, we will not need to create a fully featured `DoFn`, we can use some of the "friends" of `ParDo`.For instance, let's take some numbers as input, and multiply each number by 10:![image.png](data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAABTMAAAHyCAYAAAA3NVrNAAAgAElEQVR4Aezd+ZdUdf7n+f4z5uf5H+bnOWd6ZnqZnulZeqanv9N95vR0N8kqshSICCoq7juICirI4gYFIojigoIguwsKWG5IlVpVamlR7ttnzifre4OIyIjMjE/GjbjLg3PyZGZkfG7c+4zX5/O+7yc3Iv5J8A8BBBBAAAEEEEAAAQQQQAABBBBAAAEEECgBgX9Sgn20iwgggAACCCCAAAIIIIAAAggggAACCCCAQCAzhQABBBBAAAEEEEAAAQQQQAABBBBAAAEESkGAzCzF02QnEUAAAQQQQAABBBBAAAEEEEAAAQQQQIDMlAEEEEAAAQQQQAABBBBAAAEEEEAAAQQQKAUBMrMUT5OdRAABBBBAAAEEEEAAAQQQQAABBBBAAAEyUwYQQAABBBBAAAEEEEAAAQQQQAABBBBAoBQEyMxSPE12EgEEEEAAAQQQQAABBBBAAAEEEEAAAQTITBlAAAEEEEAAAQQQQAABBBBAAAEEEEAAgVIQIDNL8TTZSQQQQAABBBBAAAEEEEAAAQQQQAABBBAgM2UAAQQQQAABBBBAAAEEEEAAAQQQQAABBEpBgMwsxdNkJxFAAAEEEEAAAQQQQAABBBBAAAEEEECAzJQBBBBAAAEEEEAAAQQQQAABBBBAAAEEECgFATKzFE+TnUQAAQQQQAABBBBAAAEEEEAAAQQQQAABMlMGEEAAAQQQQAABBBBAAAEEEEAAAQQQQKAUBMjMUjxNdhIBBBBAAAEEEEAAAQQQQAABBBBAAAEEyEwZQAABBBBAAAEEEEAAAQQQQAABBBBAAIFSECAzS/E02UkEEEAAAQQQQAABBBBAAAEEEEAAAQQQIDNlAAEEEEAAAQQQQAABBBBAAAEEEEAAAQRKQYDMLMXTZCcRQAABBBBAAAEEEEAAAQQQQAABBBBAgMyUAQQQQAABBBBAAAEEEEAAAQQQQAABBBAoBQEysxRPk51EAAEEEEAAAQQQQAABBBBAAAEEEEAAAT<KEY>EEEEAAAQQQQAABBBBAAAEEEE<KEY>EEEEAAAQ<KEY>AQ<KEY>EEEEAAAQ<KEY>EEEEAAAQQQQA<KEY>EEEEAAAQQQ<KEY>AE<KEY>AQQQQA<KEY>EEEEAAAQQQQIDMlAEEEEAAAQQQQAABBBBAAAEEEEAAAQQQKA<KEY>EEEEAAAQQQQAABBBBAAAEyUwYQQ<KEY>EEEEAAAQQQQAABBBBAoBQEyMxSPE12EgEEEEAAAQQQQAABBBBAAAEEEEAAAQTITBlAAAEEEEAAAQQQQAABBBBAAAEEEEAAgVIQ<KEY>QQQQAABBBBAAAEEEEAAAQQQQAABBBAgM2UAAQQQQAABBBBAAAEEEEAAAQQQQAABBEpBgMwsxdNkJxFAAAEEEEAAAQQQQAABBBBAAAEEEECAzJQBBBBAAAEEEEAAAQQQQAABBBBAAAEEECgFATKzFE+TnUQAAQQQQ<KEY>EEEEAAAQQQQA<KEY>  Solution using a normal `DoFn`We need to create our own class for a `DoFn` and override the process method:<jupyter_code>class MultiplyByTenDoFn(beam.DoFn):
  pass
with beam.Pipeline(InteractiveRunner()) as p:
  pass<jupyter_output><empty_output><jupyter_text>Solution using `Map`The same problem can be simply solved with a lambda. We don't need a fully featured `DoFn` just for this task:<jupyter_code>with beam.Pipeline(InteractiveRunner()) as p:
  input = p | beam.Create([1, 2, 3, 4, 5])
  ib.show(input)

  pass<jupyter_output><empty_output><jupyter_text>The lifecycle of `DoFn`Now, in some situations we can benefit from controlling the lifecycle of our `DoFn`.In this notebook, we are going to see just how this lifecycle works.This will make much more sense when you learn about states and timers, in a later session at the Beam College. Input data for this pipeline<jupyter_code>p = beam.Pipeline(InteractiveRunner())

lines = ["En un lugar de la Mancha, de cuyo nombre no quiero acordarme,",
         "no ha mucho tiempo",
         "que vivía un hidalgo de los de lanza en astillero, adarga antigua,",
         "rocín flaco y galgo corredor."]

words = p | beam.Create(lines) \
          | beam.FlatMap(lambda x: x.split(" "))

ib.show(words)<jupyter_output><empty_output><jupyter_text>Is a `DoFn` the same as a map function?We may think that executing a `DoFn` is just the same as running a `map`.Let's see an example of a map application. Let's apply a function to sanitize the words in our `PCollection`.This function is completely stateless. It accepts a word as input, and will return a new word. There is no initialization, shutdown of any kind of connection, aggregation, state, etc. It is just a stateless map: you can pass a reference to a function or a lambda. In the above cells, we showed how to use for instance a lambda with `FlatMap`.<jupyter_code>def sanitize_word(w):
  w = w.replace(".", "").replace(",", "").replace(";", "")
  w = w.lower()
  return w
# TODO<jupyter_output><empty_output><jupyter_text>The methods of a `DoFn``You may use a `DoFn` just for a map function. For that, overwrite the `process` method and you are good to go.However, there are many more methods that you can override to control the lifecycle of a `DoFn`: creation of the worker, start of a new bundle, end of a new bundle, deletion of the worker, and of course, process of every element.The `setup` and `teardown` methods will be executed only once per worker.The `start_bundle` and `finish_bundle` will be executed once by data bundle<jupyter_code>class ChangeWordDoFn(beam.DoFn):
  pass<jupyter_output><empty_output><jupyter_text>Our `DoFn` is printing some messages at different stages in the lifecycle. In this case, we have only a handful of data, so there is only a bundle. But if we had much more data, we would see more bundles,<jupyter_code># TODO<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13938618925831203, 'fraction_numerical': 0.023817135549872123, 'mean_word_length': 6.782338308457711, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 46, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, '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': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '339379', 'license': 'Apache-2.0', 'n_tokens_mistral': 3034, 'n_tokens_neox': 2417, 'n_words': 545, 'path': 'day2/DoFn_lifecycle_[blank].ipynb', 'repo_name': 'lcaggio/beam-college', 'size': '52916'}