improves&fixes runner

Author: redvg

pkg_popularity_pipeline.py renamed to pkg_popularity_pipeline_local.py

@@ -4,9 +4,9 @@
 import apache_beam as beam
 import argparse
 
-def find_matching_lines(line, search_query):
+def find_matching_lines(line, keyword):
 
- if line.startswith(term):
+ if line.startswith(keyword):
 
  yield line
 
@@ -56,26 +56,40 @@ def compare_by_value(kv1, kv2):
 
  return value1 < value2
 
-if __name__ == '__main__':
+def run():
+
  parser = argparse.ArgumentParser(description='Find the most used Java packages')
- parser.add_argument('--output_prefix', default='/tmp/output', help='Output prefix')
- parser.add_argument('--input', default='../javahelp/src/main/java/com/google/cloud/training/dataanalyst/javahelp/', help='Input directory')
+
+ parser.add_argument('--output_prefix',
+ default='/tmp/output',
+ help='Output prefix')
+
+ parser.add_argument('--input',
+ default='../javahelp/src/main/java/com/google/cloud/training/dataanalyst/javahelp/',
+ help='Input files location directory')
 
  options, pipeline_args = parser.parse_known_args()
- p = beam.Pipeline(argv=pipeline_args)
+
+ pipeline = beam.Pipeline(argv=pipeline_args)
 
  input = '{0}*.java'.format(options.input)
+
  output_prefix = options.output_prefix
+
  keyword = 'import'
 
  # find most used packages
- (p
- | 'GetJava' >> beam.io.ReadFromText(input)
- | 'GetImports' >> beam.FlatMap(lambda line: find_matching_lines(line, keyword))
- | 'PackageUse' >> beam.FlatMap(lambda line: resolve_package_usage(line, keyword))
- | 'TotalUse' >> beam.CombinePerKey(sum)
- | 'Top_5' >> beam.transforms.combiners.Top.Of(5, compare_by_value)
- | 'write' >> beam.io.WriteToText(output_prefix)
+ (pipeline
+ | 'GetInput' >> beam.io.ReadFromText(input)
+ | 'Imports' >> beam.FlatMap(lambda line: find_matching_lines(line, keyword))
+ | 'PackageUsage' >> beam.FlatMap(lambda line: resolve_package_usage(line, keyword))
+ | 'TotalPackageUse' >> beam.CombinePerKey(sum)
+ | 'Top5PackageUsage' >> beam.transforms.combiners.Top.Of(5, compare_by_value)
+ | 'WriteOutput' >> beam.io.WriteToText(output_prefix)
  )
 
- p.run().wait_until_finish()
+ pipeline.run().wait_until_finish() 
+
+if __name__ == '__main__':
+
+ run()