main.py
def run(argv=None):
    """Main entry point; defines and runs the tfidf pipeline."""
    parser = argparse.ArgumentParser()
    parser.add_argument('--input_baseline',
                        required=False,
                        help='baseline URIs to process.')
    parser.add_argument('--input_updates',
                        required=False,
                        help='updates URIs to process.')
    parser.add_argument('--input_enriched',
                        required=False,
                        help='updates URIs to process.')
    parser.add_argument('--output',
                        required=False,
                        help='Output file to write results to.')
    parser.add_argument('--output_enriched',
                        required=False,
                        help='Output file to write results to.')
    parser.add_argument('--output_splitted',
                        required=False,
                        help='Output file to write results to.')
    known_args, pipeline_args = parser.parse_known_args(argv)
    # bq_table_schema = parse_bq_json_schema(json.load(open('schemas/medline.papers.json')))
    bq_table_schema = parse_bq_json_schema(json.loads(BQ_SCHEMA))
    # We use the save_main_session option because one or more DoFn's in this
    # workflow rely on global context (e.g., a module imported at module level).
    pipeline_options = PipelineOptions(pipeline_args)
    pipeline_options.view_as(SetupOptions).save_main_session = True
    with beam.Pipeline(options=pipeline_options) as p:

        if known_args.input_baseline or known_args.input_updates:

            if known_args.input_baseline and known_args.input_updates:
                medline_articles_base = p | 'BaselineEmitXML' >> ReadMedlineFiles(known_args.input_baseline)
                medline_articles_updates = p | 'UpdatesEmitXML' >> ReadMedlineFiles(known_args.input_updates)

                medline_articles = (
                    (medline_articles_base, medline_articles_updates)
                    | beam.Flatten())
            elif known_args.input_baseline:
                medline_articles = p | 'BaselineEmitXML' >> ReadMedlineFiles(known_args.input_baseline)
            elif known_args.input_updates:
                medline_articles = p | 'UpdatesEmitXML' >> ReadMedlineFiles(known_args.input_updates)
            else:
                raise AttributeError('at least an XML input is required')

            parsed_medline_articles = medline_articles | 'ParseXMLtoDict' >> beam.ParDo(MedlineXMLParser())

            medline_articles_grouped_by_id = parsed_medline_articles | 'GroupByPMID' >> beam.GroupByKey()

            unique_medline_articles = medline_articles_grouped_by_id | 'SortByFilename' >> beam.ParDo(
                GetLatestVersion())

            enriched_articles = unique_medline_articles | 'NLPAnalysis' >> beam.ParDo(NLPAnalysis())

            json_enriched_medline_articles = enriched_articles | 'EnrichedMedlineToJSON' >> beam.ParDo(ToJSON())

            json_enriched_medline_articles | 'WriteEnrichedJSONToGS' >> WriteToText(known_args.output_enriched,
                                                                                    file_name_suffix='_enriched.json.gz')

        elif known_args.input_enriched:

            json_enriched_medline_articles = p | 'GetEnrichedArticles' >> ReadFromText(known_args.input_enriched)

        else:
            raise AttributeError('missing json enriched data  input')

        if known_args.output_splitted:

            concepts = json_enriched_medline_articles | 'ArticleToConcepts' >> beam.ParDo(ExtractConcepts())
            concepts | 'WriteConceptJSONToGS' >> WriteToText(known_args.output_splitted,
                                                             file_name_suffix='_concepts.json.gz')

            bioentities = json_enriched_medline_articles | 'ArticleToBioentities' >> beam.ParDo(ExtractBioentities())
            bioentities | 'WriteBioentityJSONToGS' >> WriteToText(known_args.output_splitted,
                                                                  file_name_suffix='_bioentities.json.gz')

            taggedtext = json_enriched_medline_articles | 'ArticleToTaggedText' >> beam.ParDo(ExtractTaggedText())
            taggedtext | 'WriteTaggedTextJSONToGS' >> WriteToText(known_args.output_splitted,
                                                                  file_name_suffix='_taggedtext.json.gz')

            smallarticles = json_enriched_medline_articles | 'ArticleToSmallArticles' >> beam.ParDo(CleanPublication())
            smallarticles | 'WriteSmallArticleJSONToGS' >> WriteToText(known_args.output_splitted,
                                                                       file_name_suffix='_small.json.gz')
Main entry point; defines and runs the tfidf pipeline.
Main entry point; defines and runs the tfidf pipeline.
Main entry point ; defines and runs the tfidf pipeline .
def run(argv=None):
    parser = argparse.ArgumentParser()
    parser.add_argument('--input_baseline',
                        required=False,
                        help='baseline URIs to process.')
    parser.add_argument('--input_updates',
                        required=False,
                        help='updates URIs to process.')
    parser.add_argument('--input_enriched',
                        required=False,
                        help='updates URIs to process.')
    parser.add_argument('--output',
                        required=False,
                        help='Output file to write results to.')
    parser.add_argument('--output_enriched',
                        required=False,
                        help='Output file to write results to.')
    parser.add_argument('--output_splitted',
                        required=False,
                        help='Output file to write results to.')
    known_args, pipeline_args = parser.parse_known_args(argv)
    bq_table_schema = parse_bq_json_schema(json.loads(BQ_SCHEMA))
    pipeline_options = PipelineOptions(pipeline_args)
    pipeline_options.view_as(SetupOptions).save_main_session = True
    with beam.Pipeline(options=pipeline_options) as p:
        if known_args.input_baseline or known_args.input_updates:
            if known_args.input_baseline and known_args.input_updates:
                medline_articles_base = p | 'BaselineEmitXML' >> ReadMedlineFiles(known_args.input_baseline)
                medline_articles_updates = p | 'UpdatesEmitXML' >> ReadMedlineFiles(known_args.input_updates)
                medline_articles = (
                    (medline_articles_base, medline_articles_updates)
                    | beam.Flatten())
            elif known_args.input_baseline:
                medline_articles = p | 'BaselineEmitXML' >> ReadMedlineFiles(known_args.input_baseline)
            elif known_args.input_updates:
                medline_articles = p | 'UpdatesEmitXML' >> ReadMedlineFiles(known_args.input_updates)
            else:
                raise AttributeError('at least an XML input is required')
            parsed_medline_articles = medline_articles | 'ParseXMLtoDict' >> beam.ParDo(MedlineXMLParser())
            medline_articles_grouped_by_id = parsed_medline_articles | 'GroupByPMID' >> beam.GroupByKey()
            unique_medline_articles = medline_articles_grouped_by_id | 'SortByFilename' >> beam.ParDo(
                GetLatestVersion())
            enriched_articles = unique_medline_articles | 'NLPAnalysis' >> beam.ParDo(NLPAnalysis())
            json_enriched_medline_articles = enriched_articles | 'EnrichedMedlineToJSON' >> beam.ParDo(ToJSON())
            json_enriched_medline_articles | 'WriteEnrichedJSONToGS' >> WriteToText(known_args.output_enriched,
                                                                                    file_name_suffix='_enriched.json.gz')
        elif known_args.input_enriched:
            json_enriched_medline_articles = p | 'GetEnrichedArticles' >> ReadFromText(known_args.input_enriched)
        else:
            raise AttributeError('missing json enriched data  input')
        if known_args.output_splitted:
            concepts = json_enriched_medline_articles | 'ArticleToConcepts' >> beam.ParDo(ExtractConcepts())
            concepts | 'WriteConceptJSONToGS' >> WriteToText(known_args.output_splitted,
                                                             file_name_suffix='_concepts.json.gz')
            bioentities = json_enriched_medline_articles | 'ArticleToBioentities' >> beam.ParDo(ExtractBioentities())
            bioentities | 'WriteBioentityJSONToGS' >> WriteToText(known_args.output_splitted,
                                                                  file_name_suffix='_bioentities.json.gz')
            taggedtext = json_enriched_medline_articles | 'ArticleToTaggedText' >> beam.ParDo(ExtractTaggedText())
            taggedtext | 'WriteTaggedTextJSONToGS' >> WriteToText(known_args.output_splitted,
                                                                  file_name_suffix='_taggedtext.json.gz')
            smallarticles = json_enriched_medline_articles | 'ArticleToSmallArticles' >> beam.ParDo(CleanPublication())
            smallarticles | 'WriteSmallArticleJSONToGS' >> WriteToText(known_args.output_splitted,
                                                                       file_name_suffix='_small.json.gz')
def run ( argv = None ) : parser = argparse . ArgumentParser ( ) parser . add_argument ( '--input_baseline' , required = False , help = 'baseline URIs to process.' ) parser . add_argument ( '--input_updates' , required = False , help = 'updates URIs to process.' ) parser . add_argument ( '--input_enriched' , required = False , help = 'updates URIs to process.' ) parser . add_argument ( '--output' , required = False , help = 'Output file to write results to.' ) parser . add_argument ( '--output_enriched' , required = False , help = 'Output file to write results to.' ) parser . add_argument ( '--output_splitted' , required = False , help = 'Output file to write results to.' ) known_args , pipeline_args = parser . parse_known_args ( argv ) bq_table_schema = parse_bq_json_schema ( json . loads ( BQ_SCHEMA ) ) pipeline_options = PipelineOptions ( pipeline_args ) pipeline_options . view_as ( SetupOptions ) . save_main_session = True with beam . Pipeline ( options = pipeline_options ) as p : if known_args . input_baseline or known_args . input_updates : if known_args . input_baseline and known_args . input_updates : medline_articles_base = p | 'BaselineEmitXML' >> ReadMedlineFiles ( known_args . input_baseline ) medline_articles_updates = p | 'UpdatesEmitXML' >> ReadMedlineFiles ( known_args . input_updates ) medline_articles = ( ( medline_articles_base , medline_articles_updates ) | beam . Flatten ( ) ) elif known_args . input_baseline : medline_articles = p | 'BaselineEmitXML' >> ReadMedlineFiles ( known_args . input_baseline ) elif known_args . input_updates : medline_articles = p | 'UpdatesEmitXML' >> ReadMedlineFiles ( known_args . input_updates ) else : raise AttributeError ( 'at least an XML input is required' ) parsed_medline_articles = medline_articles | 'ParseXMLtoDict' >> beam . ParDo ( MedlineXMLParser ( ) ) medline_articles_grouped_by_id = parsed_medline_articles | 'GroupByPMID' >> beam . GroupByKey ( ) unique_medline_articles = medline_articles_grouped_by_id | 'SortByFilename' >> beam . ParDo ( GetLatestVersion ( ) ) enriched_articles = unique_medline_articles | 'NLPAnalysis' >> beam . ParDo ( NLPAnalysis ( ) ) json_enriched_medline_articles = enriched_articles | 'EnrichedMedlineToJSON' >> beam . ParDo ( ToJSON ( ) ) json_enriched_medline_articles | 'WriteEnrichedJSONToGS' >> WriteToText ( known_args . output_enriched , file_name_suffix = '_enriched.json.gz' ) elif known_args . input_enriched : json_enriched_medline_articles = p | 'GetEnrichedArticles' >> ReadFromText ( known_args . input_enriched ) else : raise AttributeError ( 'missing json enriched data  input' ) if known_args . output_splitted : concepts = json_enriched_medline_articles | 'ArticleToConcepts' >> beam . ParDo ( ExtractConcepts ( ) ) concepts | 'WriteConceptJSONToGS' >> WriteToText ( known_args . output_splitted , file_name_suffix = '_concepts.json.gz' ) bioentities = json_enriched_medline_articles | 'ArticleToBioentities' >> beam . ParDo ( ExtractBioentities ( ) ) bioentities | 'WriteBioentityJSONToGS' >> WriteToText ( known_args . output_splitted , file_name_suffix = '_bioentities.json.gz' ) taggedtext = json_enriched_medline_articles | 'ArticleToTaggedText' >> beam . ParDo ( ExtractTaggedText ( ) ) taggedtext | 'WriteTaggedTextJSONToGS' >> WriteToText ( known_args . output_splitted , file_name_suffix = '_taggedtext.json.gz' ) smallarticles = json_enriched_medline_articles | 'ArticleToSmallArticles' >> beam . ParDo ( CleanPublication ( ) ) smallarticles | 'WriteSmallArticleJSONToGS' >> WriteToText ( known_args . output_splitted , file_name_suffix = '_small.json.gz' )
Main entry point; defines and runs the tfidf pipeline.
Main entry point ; defines and runs the tfidf pipeline .
"""Main entry point; defines and runs the tfidf pipeline.""" # bq_table_schema = parse_bq_json_schema(json.load(open('schemas/medline.papers.json'))) # We use the save_main_session option because one or more DoFn's in this # workflow rely on global context (e.g., a module imported at module level).
argv