Build PCollection of examples in the raw (text) form.

def _build_pcollection(self, pipeline, filepaths, language):
    """Build PCollection of examples in the raw (text) form."""

    beam = tfds.core.lazy_imports.apache_beam

    def _extract_content(filepath):
      """Extracts article content from a single WikiMedia XML file."""
      logging.info("generating examples from = %s", filepath)
      with tf.io.gfile.GFile(filepath) as f:
        for _, elem in etree.iterparse(f, events=("end",)):
          if not elem.tag.endswith("page"):
            continue
          namespace = elem.tag[:-4]
          title = elem.find("./{0}title".format(namespace)).text
          ns = elem.find("./{0}ns".format(namespace)).text

          # Filter pages that are not in the "main" namespace.
          if ns != "0":
            continue

          raw_content = elem.find(
              "./{0}revision/{0}text".format(namespace)).text
          elem.clear()

          # Filter redirects.
          if raw_content is None or raw_content.lower().startswith("#redirect"):
            beam.metrics.Metrics.counter(language, "filtered-redirects").inc()
            continue

          beam.metrics.Metrics.counter(language, "extracted-examples").inc()
          yield (title, raw_content)

    def _clean_content(inputs):
      """Cleans raw wikicode to extract text."""
      title, raw_content = inputs
      try:
        text = _parse_and_clean_wikicode(raw_content)
      except (
          tfds.core.lazy_imports.mwparserfromhell.parser.ParserError) as e:
        beam.metrics.Metrics.counter(language, "parser-error").inc()
        logging.error("mwparserfromhell ParseError: %s", e)
        return

      beam.metrics.Metrics.counter(language, "cleaned-examples").inc()

      yield {
          "title": title,
          "text": text
      }

    return (
        pipeline
        | beam.Create(filepaths)
        | beam.FlatMap(_extract_content)
        | beam.FlatMap(_clean_content)
    )