Найти строку с максимальным количеством токенов с помощью Apache Beam's Python SDK - PullRequest
Новая
       8

Найти строку с максимальным количеством токенов с помощью Apache Beam's Python SDK

0 голосов
/ 10 апреля 2019

У меня есть PCollection, содержащее строки.Я хочу разбить каждую строку на пробел и найти список токенов с максимальным размером и сохранить размер в переменной как int.

.1007 *

Предложения после разбиения были бы: 

['This', 'is', 'the', 'first', 'sentence'] -> 5
['Second', 'sentence'] -> 2
['Yet', 'another', 'sentence'] -> 3

Я хочу сохранить значение 5 в переменной.

Как мне это сделать?Я наткнулся на этот пост , но он не полностью соответствует моей цели.Автор распечатывает результирующую PCollection, но я хочу использовать это значение позже на других этапах конвейера.

1 Ответ

2 голосов
/ 10 апреля 2019

Вы можете сделать это с помощью преобразования Top.of. Вкратце, мы разбиваем каждое предложение и затем вычисляем длину токена. С Top мы просто хотим получить результат номер один и передаем лямбда-функцию в качестве критерия сравнения, чтобы отсортировать их по длине слова: 

sentences = sentences = ['This is the first sentence',
       'Second sentence',
       'Yet another sentence']

longest_sentence = (p
  | 'Read Sentences' >> beam.Create(sentences)
  | 'Split into Words' >> beam.Map(lambda x: x.split(' '))
  | 'Map Token Length'      >> beam.Map(lambda x: (x, len(x)))
  | 'Top Sentence' >> combine.Top.Of(1, lambda a,b: a[1]<b[1])
  | 'Save Variable'         >> beam.ParDo(SaveMaxFn()))

, где SaveMaxFn(): 

class SaveMaxFn(beam.DoFn):
  """Stores max in global variables"""
  def process(self, element):
    length = element[0][1]
    logging.info("Longest sentence: %s tokens", length)

    return element

и length - глобальная переменная: 

global length

Результат: 

INFO:root:Longest sentence: 5 token(s)

Полный код: 

import argparse, logging

import apache_beam as beam
import apache_beam.transforms.combiners as combine
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions


class SaveMaxFn(beam.DoFn):
  """Stores max in global variables"""
  def process(self, element):
    length = element[0][1]
    logging.info("Longest sentence: %s token(s)", length)

    return element


def run(argv=None):
  parser = argparse.ArgumentParser()
  known_args, pipeline_args = parser.parse_known_args(argv)

  global length

  pipeline_options = PipelineOptions(pipeline_args)
  pipeline_options.view_as(SetupOptions).save_main_session = True
  p = beam.Pipeline(options=pipeline_options)

  sentences = sentences = ['This is the first sentence',
             'Second sentence',
             'Yet another sentence']

  longest_sentence = (p
    | 'Read Sentences' >> beam.Create(sentences)
    | 'Split into Words' >> beam.Map(lambda x: x.split(' '))
    | 'Map Token Length'      >> beam.Map(lambda x: (x, len(x)))
    | 'Top Sentence' >> combine.Top.Of(1, lambda a,b: a[1]<b[1])
    | 'Save Variable'         >> beam.ParDo(SaveMaxFn()))

  result = p.run()
  result.wait_until_finish()

if __name__ == '__main__':
  logging.getLogger().setLevel(logging.INFO)
  run()
...