def filter_rekey(element, source_id_lookup, target_id_lookup):
  """Filter merged articles and rekey by entity id."""
  key, value = element
  title, section = key

  # Only keep introductory paragraphs.
  if "INTRODUCTION" not in section:
    return

  source_text, target_text = get_text(value)
  clean_source_text, source_entities = wiki_utils.process_wikilinks(source_text)
  clean_target_text, target_entities = wiki_utils.process_wikilinks(target_text)
  updated = True
  if clean_source_text == clean_target_text:
    updated = False
  source_entities = disambiguate(source_entities, source_id_lookup)
  source_entity_ids = set(x["id"] for x in source_entities if x != title)
  target_entities = disambiguate(target_entities, target_id_lookup)
  target_entity_ids = set(x["id"] for x in target_entities if x != title)
  added_entity_ids = target_entity_ids - source_entity_ids
  added_entities = []
  for entity in target_entities:
    if entity["id"] not in added_entity_ids:
      continue
    if clean_target_text[entity["start"]:entity["end"]] in clean_source_text:
      continue
    else:
      added_entities.append(entity)
  if not added_entities:
    updated = False
  if value["source_article"]:
    value["source_article"]["text"] = clean_source_text
    value["source_article"]["entities"] = source_entities
    value["source_article"]["added_entities"] = []
  else:
    value["source_article"] = {
        "id": None,
        "ns": None,
        "text": "",
        "entities": [],
        "added_entities": []
    }
  value["target_article"]["text"] = clean_target_text
  value["target_article"]["entities"] = target_entities
  value["target_article"]["added_entities"] = added_entities
  value["updated"] = updated
  #  value["added_entity_ids"] = list(added_entity_ids)
  if updated:
    beam.metrics.Metrics.counter(__name__, "updated_articles").inc()
  else:
    beam.metrics.Metrics.counter(__name__, "unchanged_articles").inc()
  yield title, value
Filter merged articles and rekey by entity id.
language/fruit/beam_pipelines.py