def create_groups(group_ids, corpus, word, ignore_corpus, ignore_word):
  """Generate groups given the input PCollections."""

  def attach_corpus_fn(group, corpus, ignore):
    selected = None
    len_corpus = len(corpus)
    while not selected:
      c = corpus[randrange(0, len_corpus - 1)].values()[0]
      if c != ignore:
        selected = c

    yield (group, selected)

  def attach_word_fn(group, words, ignore):
    selected = None
    len_words = len(words)
    while not selected:
      c = words[randrange(0, len_words - 1)].values()[0]
      if c != ignore:
        selected = c

    yield group + (selected,)

  return (group_ids
          | df.FlatMap(
              'attach corpus',
              attach_corpus_fn,
              AsList(corpus),
              AsSingleton(ignore_corpus))
          | df.FlatMap(
              'attach word',
              attach_word_fn,
              AsIter(word),
              AsSingleton(ignore_word)))
dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/examples/cookbook/bigquery_side_input.py/create_groups/original