<jupyter_start><jupyter_text><jupyter_code>pip install apache-beam[interactive]
!pip install --quiet apache-beam
pip install apache-beam[gcp,aws,test,docs]

import apache_beam as beam

from apache_beam.options.pipeline_options import PipelineOptions

from apache_beam.io import ReadFromText
from apache_beam.io import WriteToText

p = beam.Pipeline(options=PipelineOptions())

# class GetTotal(beam.DoFn):
#     def process(self, element):
#         # get the total transactions for one item
#         return [(str(element[0]), sum(element[1]))]

data_from_source = (p
                    | 'ReadMyFile' >> ReadFromText('vgsales.csv')
                    | 'Splitter using beam.Map' >> beam.Map(lambda record: (record.split(','))[0])
                    | 'Map record to 1' >> beam.Map(lambda record: (record, 1))
                    | 'GroupBy the data' >> beam.GroupByKey()
                    | 'Get the total in each day' >> beam.ParDo(GetTotal())
                    | 'Export results to new file' >> WriteToText('Results', '.txt')
                    )

result = p.run()

# class GetTotal(beam.DoFn):
#  def process(self, element):
#  # get the total transactions for one item
# #  return [(str(element[0]),sum(element[1]))]


# vgsales = (p
#  | ‘ReadMyFile’ >> ReadFromText(‘vgsales.csv’)
#  | ‘Splitter using beam.Map’ >> beam.Map(lambda record: (record.split(‘,’))[0])
#  | ‘Map record to 1’ >> beam.Map(lambda record: (record, 1))
#  | ‘GroupBy the data’ >> beam.GroupByKey()
#  | ‘Sum using beam.Map’ >> beam.Map(lambda record: (record[0],sum(record[1])))
#  | ‘Export results to new file’ >> WriteToText( ‘Results.txt’)
#  )
# def groupbykey(test=None):
#   # [START groupbykey]
#   import apache_beam as beam

#   with beam.Pipeline() as pipeline:
#     produce_counts = (
#         pipeline
#         | beam.io.ReadFromText('vgsales.csv')
#         | 'Group counts per produce' >> beam.GroupByKey()
#         | beam.io.WriteToText(Output.txt)
# )
    
#     # [END groupbykey]
#     if test:
#       test(produce_counts)


# csv_lines = (
#     p | beam.io.ReadFromText('vgsales.csv') | 
#     beam.ParDo(Split()) |
#     beam.ParDo(CollectOpen()) |
#     "Grouping keys Open" >> beam.GroupByKey() |
    
#     ) | beam.io.WriteToText(Results.txt)
# )

! ls
! cat Results-00000-of-00001.txt
# Code to read csv file into colaboratory:
!pip install -U -q PyDrive
from pydrive.auth import GoogleAuth
from pydrive.drive import GoogleDrive
from google.colab import auth
from oauth2client.client import GoogleCredentials




# Autheticate E-Mail ID
auth.authenticate_user()
gauth = GoogleAuth()
gauth.credentials = GoogleCredentials.get_application_default()
drive = GoogleDrive(gauth)
# Get File from Drive using file-ID
# Get the file
downloaded = drive.CreateFile({'id':'1TxUK6IYJKtEYhJfHat6I-Pbpxs_qv0xc'}) # replace the id with id of file you want to access
downloaded.GetContentFile('vgsales.csv')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13595765795567316, 'fraction_numerical': 0.008931524975190208, 'mean_word_length': 3.021276595744681, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '556006', 'license': 'Apache-2.0', 'n_tokens_mistral': 1080, 'n_tokens_neox': 1022, 'n_words': 230, 'path': 'GroupByKey.ipynb', 'repo_name': 'Rajeshwari-Rudra/apache_beam-python', 'size': '3023'}