def redirect_table_pipeline(input_jsonl, output_tsv):
  """Beam pipeline for producing a redirect table as a TSV."""

  def pipeline(root):
    all_entries = (
        root
        | "ReadInput" >> beam.io.ReadFromText(input_jsonl)
        | "ParseJSONL" >> beam.Map(json.loads)
        | "CleanTitles" >> beam.Map(clean_titles))
    articles = (all_entries | "FilterArticles" >> beam.FlatMap(filter_articles))
    redirects = (
        all_entries
        | "FilterRedirects" >> beam.FlatMap(filter_redirects))
    article_id_lookup = (
        articles
        | "ArticleTitlesToIDs" >> beam.Map(lambda x: (x["title"], x["title"])))
    article_id_lookup_dict = beam.pvalue.AsDict(article_id_lookup)
    redirect_id_lookup = (
        redirects
        | "RedirectTitlesToIDs" >> beam.FlatMap(redirect_to_id,
                                                article_id_lookup_dict))
    _ = ((article_id_lookup, redirect_id_lookup)
         | "CombineTitlesToIDs" >> beam.Flatten()
         | "AsTSV" >> beam.Map("\t".join)
         | "Serialize" >> beam.io.WriteToText(output_tsv))

  return pipeline
Beam pipeline for producing a redirect table as a TSV.
language/fruit/beam_pipelines.py