-
Notifications
You must be signed in to change notification settings - Fork 2
/
filters.py
69 lines (55 loc) · 2.26 KB
/
filters.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
# python sideinput.py --runner Dataflowrunner --project $PROJECT --region us-central1 --temp_location gs://$PROJECT/tmp
from __future__ import absolute_import
import argparse
import logging
import re
#from past.builtins import unicode
import apache_beam as beam
from apache_beam.io import ReadFromText
from apache_beam.io import WriteToText
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
class TerritoryParseTuple(beam.DoFn):
def process(self, element):
territoryid, territoryname, regionid = element.split(',')
yield(int(territoryid), territoryname, int(regionid))
class FilterEven(beam.DoFn):
def process(self, element):
if element[2] % 2 == 0:
yield element
class FilterStartsWith(beam.DoFn):
def process(self, element):
if element[1].startswith('S'):
yield element
def main(argv=None, save_main_session=False):
projectid = 'qwiklabs-gcp-04-c21b49858f60'
parser = argparse.ArgumentParser()
parser.add_argument(
'--input',
dest='input',
default=f'gs://{projectid}',
help='Input file to process.')
parser.add_argument(
'--output',
dest='output',
default = f'gs://{projectid}/regions_output',
help='Output file to write results to.')
known_args, pipeline_args = parser.parse_known_args(argv)
# pipeline_options = PipelineOptions(pipeline_args, job_name=f'{projectid}-22')
pipeline_options = PipelineOptions(pipeline_args)
pipeline_options.view_as(SetupOptions).save_main_session = save_main_session
# The pipeline will be run on exiting the with block.
with beam.Pipeline(options=pipeline_options) as p:
territories = (
p | 'Read' >> ReadFromText(f'{known_args.input}/territories.csv')
| 'Parse' >> beam.ParDo(TerritoryParseTuple())
| 'Filter 1' >> beam.ParDo(FilterEven())
| 'Filter 2' >> beam.ParDo(FilterStartsWith())
# | 'Filter 1' >> beam.Filter(lambda x : x[2] % 2 == 0)
# | 'Filter 2' >> beam.Filter(lambda x : x[1].startswith('S'))
| 'Write' >> WriteToText(f'{known_args.input}/filters.csv/')
)
#p.run()
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
main()