<jupyter_start><jupyter_text>Preprocessing Pipeline NotebookDefines and deploys a preprocessing pipeline for transferring Open Addresses .csv files into parquet files formatted for NER input.<jupyter_code>import re
import apache_beam as beam
from apache_beam.options import pipeline_options
from apache_beam.options.pipeline_options import GoogleCloudOptions
from apache_beam.runners import DataflowRunner
from apache_beam.runners.interactive.interactive_runner import InteractiveRunner
import apache_beam.runners.interactive.interactive_beam as ib
from datetime import timedelta

import random
import google.auth
import pyarrow
import os
#import sys
#!{sys.executable} -m pip install pyspark
#nltk.download('averaged_perceptron_tagger')<jupyter_output><empty_output><jupyter_text>User-controlled variablesChange these variables for the data to be processed in your GCP setupInput data should always be within a single subfolder under the input location, with country folders within that.  Beyond that, files may be under state or province subfolders.  Each file should represent a single municipality, if the file is for a province or a country wide area, it should be named 'countrywide.csv', 'provincewide.csv', or 'statewide.csv'..INPUT_LOCATION  &nbsp; &nbsp; +--subfolder  &nbsp; &nbsp; &nbsp; &nbsp; +--country1  &nbsp; &nbsp; &nbsp; &nbsp; | &nbsp; +--state1  &nbsp; &nbsp; &nbsp; &nbsp; | &nbsp; | &nbsp; +--city1.csv  &nbsp; &nbsp; &nbsp; &nbsp; | &nbsp; | &nbsp; +--city2.csv  &nbsp; &nbsp; &nbsp; &nbsp; | &nbsp; +--state2  &nbsp; &nbsp; &nbsp; &nbsp; | &nbsp; | &nbsp; +--file1.csv  &nbsp; &nbsp; &nbsp; &nbsp; +-country2  &nbsp; &nbsp; &nbsp; &nbsp; | &nbsp; +--province1  &nbsp; &nbsp; &nbsp; &nbsp; | &nbsp; +--countrywide.csv  &nbsp; &nbsp; &nbsp; &nbsp; +-country3  &nbsp; &nbsp; &nbsp; &nbsp; | &nbsp; +--countrywide.csv  &nbsp; &nbsp; &nbsp; &nbsp; |-...<jupyter_code>#Bucket ID
BUCKET = 'postal-parser-28'
#If using Maplequad, should be europe-west1 or europe-west2
REGION = 'europe-west1'
INPUT_LOCATION = "gs://%s/unprocessed-data/**" % BUCKET
OUTPUT_LOCATION = "gs://%s/processed-data" % BUCKET

#Data randomization probabilities. 
PROBABILITY_SHUFFLE=0.2
PROBABILITY_DELETE=0.2
PROBABILITY_DUPLICATE=0.2

# Labels as they appear in the files being loaded (should 1:1 map to HEADER)
FILE_HEADER=['LON','LAT','NUMBER','STREET','UNIT','CITY','DISTRICT','REGION','POSTCODE','ID','HASH']
# Labels you wish to be applied in the training data
HEADER=['lon','lat','house_number','road','unit','city','state_district','state','postcode','id','hash']
PIPELINE_OPTION_FLAGS = [
    "--requirements_file=requirements.txt"
]
# If running on GCP VM, keep this as is, otherwise you can customize it to your file l
APACHE_BEAM_LOCATION = (
    '/root/apache-beam-custom/packages/beam/sdks/python/dist/apache-beam-%s0.tar.gz' % 
    beam.version.__version__)<jupyter_output><empty_output><jupyter_text>GCP SettingsSets up GCP settings, best to change settings in the above cell instead.<jupyter_code>output_training_location = "%s/training/part" % OUTPUT_LOCATION
output_testing_location = "%s/testing/part" % OUTPUT_LOCATION
dataflow_gcs_location = 'gs://%s/dataflow' % BUCKET
options = pipeline_options.PipelineOptions(PIPELINE_OPTION_FLAGS)

ib.options.capture_duration = timedelta(seconds=60)
_, options.view_as(GoogleCloudOptions).project = google.auth.default()
options.view_as(GoogleCloudOptions).region = REGION
options.view_as(pipeline_options.SetupOptions).sdk_location = APACHE_BEAM_LOCATION
options.view_as(GoogleCloudOptions).staging_location = '%s/staging' % dataflow_gcs_location
options.view_as(GoogleCloudOptions).temp_location = '%s/temp' % dataflow_gcs_location<jupyter_output><empty_output><jupyter_text>Address ClassRepresents a single address and facilitates the transition from structured address to free-form string.  Also assigns proper NER tags to each entity. Executed after first 2 Apache Beam PTransforms. ParDo(AddressFunc()) Output PCollection:  | 'text'                  | 'labels'                                                | 'tokens'                          |   |   ||-------------------------|---------------------------------------------------------|-----------------------------------|---|---|| '70 York St Toronto ON' | ['B-house_number','B-street','I-street','city','state'] | ['70','York','St','Toronto','ON'] |   |   || ...                     | ...                                                     | ...                               |   |   ||                         |                                                         |                                   |   |   |<jupyter_code>class AddressFunc(beam.DoFn):
    def init(self):
        probability_shuffle=0.2 
        probability_delete=0.2 
        probability_duplicate=0.2
        import random 
        
        #import nltk
        #nltk.download('averaged_perceptron_tagger')
        
        self.MAX_NUM_TAGS = 20
        self._set_order()
        if random.random() < probability_shuffle:
            self._randomize_order()
        self._duplicate_tags(probability_duplicate)
        self._delete_tags(probability_delete)
        self.ordered = False
        
    def _set_order(self):
        import random
        r = random.randint(0,2)
        if r == 0:
                new_order = ['house_number', 'road', 'city', 'city_district',
                         'state_district', 'state', 'postcode', 'house', 'level', 'unit', 'po_box',
                         'country']
        elif r == 1:
            new_order = ['house', 'house_number', 'po_box', 'road', 'city',
                         'city_district', 'state_district', 'state', 'postcode', 'level', 'unit',
                         'country']
        else:
            new_order = ['house', 'level', 'unit', 'po_box', 'house_number',
                         'road', 'city', 'city_district',
                         'state_district', 'state', 'postcode', 'country']
        self.ordered = False
        self.order = new_order

    def _randomize_order(self):
        import random
        random.shuffle(self.order)
        self.ordered = False

    def _delete_tags(self, _delete_probability):
        import random
        while random.random() < _delete_probability and len(self.order) > 1:
            del(self.order[random.randint(0, len(self.order)-1)])
        self.ordered = False

    def _duplicate_tags(self, _duplicate_probability):
        import random
        while random.random() < _duplicate_probability and len(self.order) < self.MAX_NUM_TAGS:
            item_to_be_duplicated = self.order[random.randint(0, len(self.order)-1)]
            self.order.insert(random.randint(0, len(self.order)), item_to_be_duplicated)
        self.ordered = False
    
    def _remove_extra_labels(self):
        # Description: Sorts csv_dict to create a list of dictionaries
        #   such that they are in the same order they would be in an
        #   address string written by a human.  Uses the order stored in class
        i = 0
        while i < len(self.order):
            if not self.order[i] in self.address_dict:
                del(self.order[i])
            else:
                i += 1
        self.ordered = True

    def _ner_tags(self):
        tags = []
        for header in self.order:
            value = re.split('[ _]',self.address_dict[header])
            tokens = []
            tokens = tokens + [word for word in value if word]
            for i in range(len(tokens)):
                if i == 0:
                    tags.append('B-' + header)
                else:
                    tags.append('I-' + header)
        return tags

    def _tokenize(self):
        tokens = []
        for header in self.order:
            value = re.split('[ _]', self.address_dict[header])
            tokens = tokens + [word for word in value if word]
        return tokens
    
    def _to_string(self):
        string_representation = ''
        for k in self.order:
            string_representation += self.address_dict[k] + ' '
        string_representation = string_representation.strip()
        return string_representation
    
    def _label(self):
        if not self.ordered:
            self._remove_extra_labels()
        return {'text': self._to_string(), 'labels': self._ner_tags() ,'tokens': self._tokenize()}
    
    def process(self, element):
        self.address_dict = element
        self.init()
        return [self._label()]<jupyter_output><empty_output><jupyter_text>Apache Beam PTransform and DoFn classesCalled throughout the pipeline execution  FileReader Output PCollection| 'filename'            | 'row'                                                        |   |   |   ||-----------------------|--------------------------------------------------------------|---|---|---|| ['ca','on','Toronto'] | '11.521,47.2313,70,York St,,Toronto,,,M5H 1J8,,some_hash_id' |   |   |   || ...                   | ...                                                          |   |   |   ||                       |                                                              |   |   |   |<jupyter_code>class FileReader(beam.PTransform):
    # Traverses GCP Storage and reads in any objects with prefix FILE_LOCATION and suffix '.csv'
    # Flattens all file reads into a single PCollection
    def __init__(self, FILE_LOCATION, pipeline_options):
        self._FILE_LOCATION = FILE_LOCATION
        self._options = pipeline_options
    
    def expand(self, pcoll):
        from apache_beam.io.gcp.gcsfilesystem import GCSFileSystem
        from apache_beam.io.filesystem import FileSystem
        from apache_beam.io.filesystem import FileMetadata
        from operator import add
        from functools import reduce
        import re
        
        gcs = GCSFileSystem(self._options)
        result = [m.metadata_list for m in gcs.match([self._FILE_LOCATION])]
        result = reduce(add, result)
        variables = ['p{}'.format(i) for i in range(len(result))]
        read_labels = ['Read file {}'.format(i) for i in range(len(result))]
        add_filename_labels = ['Add filename {}'.format(i) for i in range(len(result))]
        
        return (
                [pcoll.pipeline 
                    | read_labels[i] >> beam.io.ReadFromText(result[i].path, skip_header_lines=1) 
                    | add_filename_labels[i] >> beam.ParDo(AddFilenamesFn(), result[i].path, self._FILE_LOCATION) 
                    for i in range(len(result)) if result[i].path.endswith('.csv')]
                | 'Flatten PCollections' >> beam.Flatten() 
            )

class AddFilenamesFn(beam.DoFn):
    # ParDo to output a dict with filename and row
    def process(self, element, file_path, base_path):
        file_path_list = file_path.replace('.csv','').replace('_',' ').split("/")[len(base_path.split('/'))-1:]
        yield {'filename':file_path_list, 'row':element}<jupyter_output><empty_output><jupyter_text>InputFormatting Output PCollection (Stored as collection of dictionaries)| 'lat'    | 'lon'     | 'house_number' | 'road'    | 'state' | 'city'    | 'postcode' | ... ||----------|-----------|----------------|-----------|---------|-----------|------------|-----|| '11.521' | '47.2313' | '70'           | 'York St' | 'on'    | 'Toronto' | 'M5H 1J8'  | ... || ...      | ...       | ...            | ...       | ...     | ...       | ...        | ... ||          |           |                |           |         |           |            |     |<jupyter_code>class InputFormatting(beam.PTransform):
    # Basic file formatting to extract data from files
    def __init__(self, header):
        self._header = header
    def expand(self, pcoll):
        return (
            pcoll
            | 'Parse CSV' >> beam.ParDo(ParseCSV())
            | 'Build Dictionary' >> beam.ParDo(ToDict(), self._header)
        )
        
class ParseCSV(beam.DoFn):    
    # Parses out a line of text as a csv line
    def process(self, element):
        import csv
        for line in csv.reader([element['row'].replace('.','')], quotechar='"', delimiter=',', quoting=csv.QUOTE_ALL, skipinitialspace=True):
            return [{'filename':element['filename'], 'row':line}]
        
class ToDict(beam.DoFn):
    # Breaks down a row to be a dictionary with header : entity
    def process(self, element, header):
        import re
        
        #add info from the filename to the dictionary
        FILE_NAME_FIELDS=['country','state']
        line_dict = {}
        for i in range(min(len(element['filename'])-2, 2)):
            line_dict[FILE_NAME_FIELDS[i]] = element['filename'][i+1]
        file_name = element['filename'][-1]
        if not re.match('.*([0-9]|country|province|state|wide).*', file_name):
            line_dict['city'] = file_name
        
        #insert the info from the row itself into the dictionary
        for i in range(min(len(header),len(element['row']))):
            #TODO: Have entity headers be based on the header in the file rather than predefined
            val = element['row'][i]
            if val != '':
                line_dict[header[i]] = element['row'][i]
        return([line_dict])<jupyter_output><empty_output><jupyter_text>NERFormatting Output PCollection| 'text'                  | 'label'                                                                                                                        |   |   |   ||-------------------------|--------------------------------------------------------------------------------------------------------------------------------|---|---|---|| '70 York St Toronto ON' | [{'annotatorType':'named_entity','begin':0,'end':1,'result':'B-house_number','metadata':{'word':'70'},'embeddings':[0.0]},...] |   |   |   || ...                     | ...                                                                                                                            |   |   |   ||                         |                                                                                                                                |   |   |   |<jupyter_code>class NERFormatting(beam.PTransform):
    #Formats the PCollection into a format to be taken in by the NER model
    def expand(self, pcoll):
        return (
            pcoll
            | 'To NER Inout' >> beam.ParDo(ToNER())
            | 'Partition' >> beam.Partition(self.by_random, 2)
        )
    def by_random(self, element, num_partitions):
        import random
        PERCENT_TRAINING_DATA = 0.8
        if random.random() < PERCENT_TRAINING_DATA:
            return 0
        return 1

class ToNER(beam.DoFn):
    #Changes the list of tokens and labels to the format needed for NER input for a given address
    #element: {'tokens': [5, Main, St., ...], 'labels': [B-house_number, B-road, I-road]}
    def process(self, element):
        tokens, labels = element['tokens'], element['labels']
        data=[]
        lastBegin = 0
        for i in range(len(tokens)):
            a = {
                'annotatorType' :  "named_entity",
                'begin' : lastBegin,
                'end' : lastBegin + len(tokens[i]) - 1,
                'result' : labels[i],
                'metadata' :  {'word': tokens[i]},
                'embeddings' : [0.00]
            }
            lastBegin += len(tokens[i])+1
            data.append(a)
        return [{'text':element['text'], 'label':data}]<jupyter_output><empty_output><jupyter_text>Pipeline DeclarationDefines the path through which the pipeline executes<jupyter_code>p = beam.Pipeline(InteractiveRunner())

train, test = (
    p 
    | 'Read files' >> FileReader(INPUT_LOCATION, options)
    | 'Format Input' >> InputFormatting(HEADER)
    | 'Order Adress' >> beam.ParDo(AddressFunc())
    | 'To NER Input' >> NERFormatting()
    )
# Outputs training split of data to training parquet
_ = train | 'Write train parquet' >> beam.io.parquetio.WriteToParquet(
        file_path_prefix=output_training_location,
        schema=pyarrow.schema(
            [('text',pyarrow.string()),
             ('label',pyarrow.list_(
                 pyarrow.struct([
                    ('annotatorType', pyarrow.string()),
                    ('begin', pyarrow.int32()),
                    ('end', pyarrow.int32()),
                    ('result', pyarrow.string()),
                    ('metadata',pyarrow.struct([('word', pyarrow.string())])),
                    ('embeddings', pyarrow.list_(pyarrow.float64()))
                     ])))]
        ),
        file_name_suffix='.parquet')

#Outputs testing split of data to testing parquet
_ = test | 'Write test parquet' >> beam.io.parquetio.WriteToParquet(
        file_path_prefix=output_testing_location,
        schema=pyarrow.schema(
            [('text',pyarrow.string()),
             ('label',pyarrow.list_(
                 pyarrow.struct([
                    ('annotatorType', pyarrow.string()),
                    ('begin', pyarrow.int32()),
                    ('end', pyarrow.int32()),
                    ('result', pyarrow.string()),
                    ('metadata',pyarrow.struct([('word', pyarrow.string())])),
                    ('embeddings', pyarrow.list_(pyarrow.float64()))
                     ])))]
        ),
        file_name_suffix='.parquet')<jupyter_output><empty_output><jupyter_text>Run pipeline on Dataflow<jupyter_code>ib.show_graph(p)
pipeline_result = DataflowRunner().run_pipeline(p, options=options)
from IPython.core.display import display, HTML
url = ('https://console.cloud.google.com/dataflow/jobs/%s/%s?project=%s' % 
      (pipeline_result._job.location, pipeline_result._job.id, pipeline_result._job.projectId))
display(HTML('Click <a href="%s" target="_new">here</a> for the details of your Dataflow job!' % url))<jupyter_output><empty_output><jupyter_text>Show PCollection for testing purposesRuns locally, don't run this on large datasets<jupyter_code>#ib.show(train)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.15047650894499248, 'fraction_numerical': 0.006855040963049657, 'mean_word_length': 2.0710251583090877, 'pattern_counts': {'":': 0, '<': 49, '<?xml version=': 0, '>': 70, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 56, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '474854', 'license': 'MIT', 'n_tokens_mistral': 5191, 'n_tokens_neox': 4997, 'n_words': 1172, 'path': 'preprocessing-pipeline/preprocessing-pipeline.ipynb', 'repo_name': 'navh/postal-parser', 'size': '17943'}