더블리의 12층

sentence embedding을 얻었다면, 두개의 유사도를 계산하기 위해서는 cosine similarity를 이용해서 계산을 해야한다.

pandas에서 udf를 통해 계산하는 방법은 아래와 같다.
keyword와 context에는 문자열이 들어가면 된다.

ex: keyword: 안녕, context: 잘가요. 멀리 안가요

import numpy as np
from scipy import spatial

def sim(x, y): 
  embed1 = get_embed(x)  
  embed2 = get_embed(y) 
  return 1 - spatial.distance.cosine(embed1, embed2)

def sim_udf(x): 
  sim_value = sim(x['keyword'], x['context'])  
  return sim_value 


df['cosim'] = df.apply(sim_udf, axis=1)

DataFrame GroupBy를 하고 apply를 통해 내가 지정한 함수로 계산이 가능하도록 구현이 가능하다. 보통은 groupby('columnname').sum()을 통해 특정 컬럼에 대해서 그룹별로 합, 평균, 편차 등을 계산하게 된다. 조금더 복잡한 계산을 할 수 있을까?

여기서 복잡하다는 말은 특정 컬럼은 그룹별로 string의 리스트로 반환될 수 있다. 숫자의 경우 단순히 합을 계산하거나 평균을 계산하면 되지만, string은 concat을 하거나, dict, list의 형태로 반환을 해야하는 경우가 있다.

def f(x):
  x = x.C.unique()[0] # C컬럼에서 unique한 값을 빼내고 첫번째 값으로 (groupby로 포함해도 상관없음)
  return pd.Series(dict(A = x['A'].sum(), 
                    B = x['B'].sum(), 
                    C = "{%s}" % ', '.join(x['C'])))

df.groupby('A').apply(f)
"""
A         B               C
A                             
1  2  1.615586  {This, string}
2  4  0.421821         {is, !}
3  3  0.463468             {a}
4  4  0.643961        {random}
"""

pandas를 사용하다보면 여러개의 컬럼의 결과를 하나의 값으로 계산할때도 있지만, 여러개의 값으로 여러개의 값을 계산하고 싶을때가 있다. 이때는 아래와 같이 하면 multiple columns의 결과를 받을 수 있다.

udf에서 두개의 값을 반환한다면, df에서 각각의 컬럼에 대해서 반환값을 넣어주고, zip(*df.apply)를 해줘야 두개의 컬럼으로 각각 값이 들어간다

def preprocessing_udf(x):  
  keyword = preprocessing(x['keyword'])
  context = preprocessing(x['context'])
  return keyword, context

def parallel_preprocessing(df): 
  # df['pre_context'] = df.progress_apply(preprocessing_udf, axis=1)
  df['pre_keyword'], df['pre_context'] = zip(*df.apply(preprocessing_udf, axis=1))
  return df

parallelize_dataframe(result_sample_df, parallel_preprocessing,n_cores=2)

• dataframe에서 apply의 수행시간이 매우 느리다면 여러개의 프로세스를 사용해 처리하는 방법을 고려해보는게 좋다.
• multiprocessing.cpu_count()을 통해 cpu 코어의 수를 가져올 수 있다.
• 코어수를 가져오고 udf를 생성했다면 아래와 같이 구현하면 된다.

def text_preprocessing(df): 
  # df['pre_context'] = df.progress_apply(preprocessing_udf, axis=1) # 다음과같이 진행상황을 출력하고 싶었으나 화면에 이상하게 출력됨
  df['p_context'] = df.apply(preprocessing_udf, axis=1)
  return df

from multiprocessing import  Pool
import multiprocessing
print ('cpu counts: %d' % multiprocessing.cpu_count())

def parallelize_dataframe(df, func, n_cores=8):
    df_split = np.array_split(df, n_cores) # core의 개수로 df를 나눈다. 
    pool = Pool(n_cores) # pool을 cpu 코어 개수 만큼 생성하고
    df = pd.concat(pool.map(func, df_split)) # 나눠진 df를 func으로 넘겨서 수행한다.
    pool.close()
    pool.join() # 모두가 완료될때까지 대기
    return df

result_sample = parallelize_dataframe(result_df, text_preprocessing)

• 코어의 개수에따라 linear하게 성능이 증가하진 않지만 상당한 성능 향상에 기여할 수 있음
• 참고
  • https://towardsdatascience.com/make-your-own-super-pandas-using-multiproc-1c04f41944a1

• pandas에서 apply함수를 통해 하나의 row를 처리할때 시간이 오래 걸리는 경우, 어느정도 얼마나 처리가 되었는지 확인이 어렵다
• apply의 진행상황을 가져오기 위해서 tqdm을 사용해보자
• tqdm을 사용하면 아래와 같이 진행상황을 알려준다.

 0%|          | 39/10000 [00:13<58:15,  2.85it/s]  

• tqdm을 사용하면 수행해야 하는 row의 개수와 row 하나를 처리하는데 걸리는 수행시간을 함께 알려주기 때문에 속도개선을 하는데도 도움을 줄 수 있다. (row한번 작업하는데 얼마나 수행시간이 걸리는지 확인하면서 최적화)

import pandas as pd
import numpy as np
from tqdm import tqdm
# from tqdm.auto import tqdm  # for notebooks

df = pd.DataFrame(np.random.randint(0, int(1e8), (10000, 1000)))

# Create and register a new `tqdm` instance with `pandas`
# (can use tqdm_gui, optional kwargs, etc.)
tqdm.pandas()

# Now you can use `progress_apply` instead of `apply`
df.groupby(0).progress_apply(lambda x: x**2)

df_users.groupby(['userID', 'requestDate']).apply(feature_rollup)

from tqdm import tqdm
tqdm.pandas()
df_users.groupby(['userID', 'requestDate']).progress_apply(feature_rollup)

Note: tqdm <= v4.8 에서는 tqdm.pandas() 대신에 아래와 같이 사용해야 한다.

from tqdm import tqdm, tqdm_pandas
tqdm_pandas(tqdm())

• 참고
  • https://stackoverflow.com/questions/18603270/progress-indicator-during-pandas-operations

pandas에서 데이터를 읽을때 특정조건을 필터할필요가 있다. 전체를 읽고 필터링하기 보다는 내가 필요한 데이터만 읽고 필터링을 하자! 읽으면서 filtering을 하기 위해서는 chunksize를 사용해야 한다.

import pandas as pd
iter_csv = pd.read_csv('file.csv', iterator=True, chunksize=1000)
df = pd.concat([chunk[chunk['field'] > constant] for chunk in iter_csv])

def read_result():
  lines = open('/tmp/query_result.tsv', 'r').readlines()
  data = []
  cols = lines[0][:-1].split("\t")
  len_cols = len(cols)

  for line in lines[1:]:
    vals = line[:-1].split("\t")  
    if len(vals) != len_cols:
      # print (line[:-1])
      continue
    data.append(vals)  
  return pd.DataFrame(data, columns=cols)

jupyter notebook --generate-config의 명령을 통해 기본 경로 ~/.jupyter에 config파일을 생성할수 있다.

만약 이미 생성이 되어 있다면 jupyter --config-dir을 통해 경로를 확인할 수 있다. 경로를 확인하고 아래 코드를 통해 환경변수를 추가하자

import os
c = get_config()
os.environ['LD_LIBRARY_PATH'] = '/home1/jslee/library/lib'
os.environ['PYTHONPATH'] = '${PYTHONPATH}:/home1/jslee/library/binding/python'

c.Spawner.env.update('LD_LIBRARY_PATH')
c.Spawner.env.update('PYTHONPATH')

• 관련이슈
  • https://github.com/jupyter/notebook/issues/1290

Ipython, JupyterNotebook을 사용하다보면,
df.head(100)의 결과를 출력할 경우가 있다. (하지만? 10개정도 보일것이다.)
df.head(1)의 결과를 출력하니 컬럼에 ...으로 나올때도 있다.
df.head(1)의 결과에서 dataframe의 폭이 좁을때가 있다.

이런 여러가지 상황에서 dataframe의 출력 결과를 설정하는게 필요하다.
아래 pd.option.display 또는 pd.set_option을 통해 변경이 가능하다.

pd.options.display.max_columns = 30
pd.options.display.max_rows = 20

pd.set_option('display.height', 1000)
pd.set_option('display.max_rows', 200)
pd.set_option('display.max_columns', 500)
pd.set_option('display.width', 1000)