как распараллелить части функции тегов речи в блоках данных (python / pyspark) - PullRequest
Новая
       7

как распараллелить части функции тегов речи в блоках данных (python / pyspark)

0 голосов
/ 07 февраля 2020

Прямо сейчас у меня есть ~ 5000 строк существенных текстовых данных, которые мне нужны для пометки POS. Сначала я попытался с помощью стандартного python кода сделать это, но время выполнения составило более 2 часов (см. Код ниже) 

def pos_counts(page):
 '''returns common pos counts for a single block of text - ['VERB', 'NOUN', 'ADJ', 'ADV', 'NUM', '.', 'DET']
 parameters:
   page (str) : block of text
 returns:
   l (dict) : dictionary where the keys are the pos tags and the values are the count of that tag in page
 '''
 words = nltk.word_tokenize(page)
 page_tagged = nltk.tag.pos_tag(words, tagset='universal')
 l = dict(Counter(pd.DataFrame([{'pos':key, 'word':val} for val, key in page_tagged])['pos'].values))
 l['Text'] = page
 return l

pos = [] # list to store each of the dictionaries returned by pos_counts()

for text in df['Text'].values:
 pos.append(pos_counts(text))

pos_df = pd.DataFrame(pos)

После этого я попытался распараллелить его с помощью pandas_udf: 

from pyspark.sql.functions import col, pandas_udf
from pyspark.sql.types import IntegerType

def noun_pos_counts(pages):
  '''gives the noun count for each page in pages
  parameters:
    pages (pandas series) : series with each row containing a text block
  returns:
    result (pandas series) : series with the noun count in each block of text
  '''
  noun_counts = []
  for page in pages:
    page = str(page)
    words = nltk.word_tokenize(page)
    page_tagged = nltk.tag.pos_tag(words, tagset='universal')
    l = dict(Counter(pd.DataFrame([{'pos':key, 'word':val} for val, key in page_tagged])['pos'].values))
    noun_counts.append(l['NOUN'])
  return pd.Series(noun_counts)

noun_udf = pandas_udf(noun_pos_counts, returnType=IntegerType())

sparkdf = spark.createDataFrame(pd.DataFrame(df['Text'].fillna("None"), columns=["Text"]))
pos_df = sparkdf.select(noun_udf(col("Text")).alias("NOUN")).toPandas()

Этот второй метод занял еще больше времени (~ 3,5 часа) строго из-за вызова toPandas() в конце.

Есть ли другой способ сделать это со значительно более коротким временем выполнения?

...