<jupyter_start><jupyter_text>Pipeline - Using Map step by step<jupyter_code>import apache_beam as beam

p1 = beam.Pipeline()

# Read / Write pipeline:
attendance_count = (
    p1
    |beam.io.ReadFromText('data/dept_data.txt')
    
    |beam.io.WriteToText('output/output_map1')
)

p1.run()

# visualize output
!{('head -n 10 output/output_map1-00000-of-00001')}
import apache_beam as beam

p1 = beam.Pipeline()

# Split the data:
attendance_count = (
    p1
    |beam.io.ReadFromText('data/dept_data.txt')
    |beam.Map(lambda record: record.split(','))
    |beam.io.WriteToText('output/output_split')
  
)

p1.run()

# visualize output
!{('head -n 10 output/output_split-00000-of-00001')}
import apache_beam as beam

p1 = beam.Pipeline()

# Split + filter the data:
attendance_count = (
    p1
    |beam.io.ReadFromText('data/dept_data.txt')
    |beam.Map(lambda record: record.split(','))
    |beam.Filter(lambda record: record[3] == 'Accounts')
    |beam.io.WriteToText('output/output_filter')
  
)

p1.run()

# visualize output
!{('head -n 20 output/output_filter-00000-of-00001')}
import apache_beam as beam

p1 = beam.Pipeline()

# Split + filter + key-value:
attendance_count = (
    p1
    |beam.io.ReadFromText('data/dept_data.txt')
    |beam.Map(lambda record: record.split(','))
    |beam.Filter(lambda record: record[3] == 'Accounts')
    |beam.Map(lambda record: (record[1], 1))
    |beam.io.WriteToText('output/output_kv')
  
)

p1.run()

# visualize output
!{('head -n 10 output/output_kv-00000-of-00001')}
import apache_beam as beam

with beam.Pipeline() as pipe:
    # Split + filter + key-value + Sum by key:
    attendance_count = (
        pipe
        |"Read file" >> beam.io.ReadFromText('data/dept_data.txt')
        |"Split"     >> beam.Map(lambda record: record.split(','))
        |"Filter"    >> beam.Filter(lambda record: record[3] == 'Accounts')
        |"KV"        >> beam.Map(lambda record: (record[1], 1))     # |beam.Map(lambda record: (record[1], 1))  # By ID
        |"Sum"       >> beam.CombinePerKey(sum)   # GropuBy + Combiner + Reducer
        |"Output"    >> beam.io.WriteToText('output/output_sum')
    )

# visualize output
!{('head -n 10 output/output_sum-00000-of-00001')}<jupyter_output>WARNING:apache_beam.io.filebasedsink:Deleting 1 existing files in target path matching: -*-of-%(num_shards)05d
{'fraction_non_alphanumeric': 0.16083617747440274, 'fraction_numerical': 0.03668941979522184, 'mean_word_length': 3.5711500974658867, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 16, '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': '392088', 'license': 'MIT', 'n_tokens_mistral': 948, 'n_tokens_neox': 863, 'n_words': 167, 'path': 'apache_beam_02_IO.ipynb', 'repo_name': 'avallarino-ar/apache_beal_lab', 'size': '2344'}