mirror of
https://github.com/dtomlinson91/street_group_tech_test
synced 2025-12-22 03:55:43 +00:00
adding latest beam pipeline code
This commit is contained in:
@@ -17,7 +17,7 @@ def csv_reader(csv_file):
|
|||||||
return csv.reader(io.TextIOWrapper(csv_file.open()))
|
return csv.reader(io.TextIOWrapper(csv_file.open()))
|
||||||
|
|
||||||
|
|
||||||
def slicer(element, *ranges):
|
def slice_by_range(element, *ranges):
|
||||||
return chain(*(islice(element, *r) for r in ranges))
|
return chain(*(islice(element, *r) for r in ranges))
|
||||||
|
|
||||||
|
|
||||||
@@ -27,12 +27,18 @@ class SplitPAON(beam.DoFn):
|
|||||||
|
|
||||||
if len(paon_split) == 0:
|
if len(paon_split) == 0:
|
||||||
return element
|
return element
|
||||||
elif len(paon_split) == 1:
|
elif len(paon_split) > 0:
|
||||||
pass
|
|
||||||
else:
|
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class DropEmptyColumn(bean.doFn):
|
||||||
|
def __init__(self, index):
|
||||||
|
self.index = index
|
||||||
|
|
||||||
|
def process(self, element):
|
||||||
|
column = element[self.index]
|
||||||
|
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
csv_data = resources.path(
|
csv_data = resources.path(
|
||||||
"analyse_properties.data", "pp-monthly-update-new-version.csv"
|
"analyse_properties.data", "pp-monthly-update-new-version.csv"
|
||||||
@@ -53,7 +59,7 @@ def main():
|
|||||||
clean = (
|
clean = (
|
||||||
load
|
load
|
||||||
| "Drop unneeded columns"
|
| "Drop unneeded columns"
|
||||||
>> beam.Map(lambda element: list(slicer(element, (1, 4), (7, 14))))
|
>> beam.Map(lambda element: list(slice_by_range(element, (1, 4), (7, 14))))
|
||||||
| "Convert to Upper Case"
|
| "Convert to Upper Case"
|
||||||
>> beam.Map(lambda element: [e.upper() for e in element])
|
>> beam.Map(lambda element: [e.upper() for e in element])
|
||||||
| beam.Map(print)
|
| beam.Map(print)
|
||||||
|
|||||||
Reference in New Issue
Block a user