どのくらいの環境のコピーがスパークしますか？

Question

私は約5GBの圧縮データ（文字列）を精緻化しなければならないPySparkアプリケーションを持っています。私は12コア（24スレッド）と72GbのRAMを搭載した小型サーバーを使用しています。私のPySparkプログラムは2つのマップ操作のみで構成されており、3つの非常に大きな正規表現（各3GBがすでにコンパイルされている）とpickleがロードされています。 Sparkは、同じマシン上のワーカーとマスターとのスタンドアロンモードで作業しています。

私の質問です：各エグゼクティブコアの各変数をsparkレプリケートしますか？使用可能なすべてのメモリを使用し、多くのスワップ領域を使用するためです。それともRAM内のすべてのパーティションをロードするのでしょうか？ RDDには、3つの正規表現によって検索されなければならない約10百万の文字列が含まれています。 RDDは約1000のパーティションをカウントします。私はこの作業を完了するのに困っています。何分後にメモリがいっぱいになり、スワップスペースが非常に遅くなってしまいます。 私は正規表現なしでは状況は同じであることに気付きました。

これは、それがTwitterのつぶやきのすべての無用のフィールドを削除し、特定の単語をつぶやきのテキストや説明をスキャンし、私のコードです：

import json 
import re 
import twitter_util as twu 
import pickle 

from pyspark import SparkContext 
sc = SparkContext() 

prefix = '/home/lucadiliello' 

source = prefix + '/data/tweets' 
dest = prefix + '/data/complete_tweets' 

#Regex's path 
companies_names_regex = prefix + '/data/comp_names_regex' 
companies_names_dict = prefix + '/data/comp_names_dict' 
companies_names_dict_to_legal = prefix + '/data/comp_names_dict_to_legal' 

#Loading the regex's 
comp_regex = pickle.load(open(companies_names_regex)) 
comp_dict = pickle.load(open(companies_names_dict)) 
comp_dict_legal = pickle.load(open(companies_names_dict_to_legal)) 

#Loading the RDD from textfile 
tx = sc.textFile(source).map(lambda a: json.loads(a)) 


def get_device(input_text): 
    output_text = re.sub('<[^>]*>', '', input_text) 
    return output_text 

def filter_data(a): 
    res = {} 
    try: 
     res['mentions'] = a['entities']['user_mentions'] 
     res['hashtags'] = a['entities']['hashtags'] 
     res['created_at'] = a['created_at'] 
     res['id'] = a['id'] 

     res['lang'] = a['lang'] 
     if 'place' in a and a['place'] is not None:  
      res['place'] = {} 
      res['place']['country_code'] = a['place']['country_code'] 
      res['place']['place_type'] = a['place']['place_type'] 
      res['place']['name'] = a['place']['name'] 
      res['place']['full_name'] = a['place']['full_name'] 

     res['source'] = get_device(a['source']) 
     res['text'] = a['text'] 
     res['timestamp_ms'] = a['timestamp_ms'] 

     res['user'] = {} 
     res['user']['created_at'] = a['user']['created_at'] 
     res['user']['description'] = a['user']['description'] 
     res['user']['followers_count'] = a['user']['followers_count'] 
     res['user']['friends_count'] = a['user']['friends_count'] 
     res['user']['screen_name'] = a['user']['screen_name'] 
     res['user']['lang'] = a['user']['lang'] 
     res['user']['name'] = a['user']['name'] 
     res['user']['location'] = a['user']['location'] 
     res['user']['statuses_count'] = a['user']['statuses_count'] 
     res['user']['verified'] = a['user']['verified'] 
     res['user']['url'] = a['user']['url'] 
    except KeyError: 
     return [] 

    return [res] 


results = tx.flatMap(filter_data) 


def setting_tweet(tweet): 

    text = tweet['text'] if tweet['text'] is not None else '' 
    descr = tweet['user']['description'] if tweet['user']['description'] is not None else '' 
    del tweet['text'] 
    del tweet['user']['description'] 

    tweet['text'] = {} 
    tweet['user']['description'] = {} 
    del tweet['mentions'] 

    #tweet 
    tweet['text']['original_text'] = text 
    tweet['text']['mentions'] = twu.find_retweet(text) 
    tweet['text']['links'] = [] 
    for j in twu.find_links(text): 
     tmp = {} 
     try: 
      tmp['host'] = twu.get_host(j) 
      tmp['link'] = j 
      tweet['text']['links'].append(tmp) 
     except ValueError: 
      pass 

    tweet['text']['companies'] = [] 
    for x in comp_regex.findall(text.lower()): 
     tmp = {} 
     tmp['id'] = comp_dict[x.lower()] 
     tmp['name'] = x 
     tmp['legalName'] = comp_dict_legal[x.lower()] 
     tweet['text']['companies'].append(tmp) 

    # descr 
    tweet['user']['description']['original_text'] = descr 
    tweet['user']['description']['mentions'] = twu.find_retweet(descr) 
    tweet['user']['description']['links'] = [] 
    for j in twu.find_links(descr): 
     tmp = {} 
     try: 
      tmp['host'] = twu.get_host(j) 
      tmp['link'] = j 
      tweet['user']['description']['links'].append(tmp) 
     except ValueError: 
      pass 

    tweet['user']['description']['companies'] = [] 
    for x in comp_regex.findall(descr.lower()): 
     tmp = {} 
     tmp['id'] = comp_dict[x.lower()] 
     tmp['name'] = x 
     tmp['legalName'] = comp_dict_legal[x.lower()] 
     tweet['user']['description']['companies'].append(tmp) 

    return tweet 


res = results.map(setting_tweet) 

res.map(lambda a: json.dumps(a)).saveAsTextFile(dest, compressionCodecClass="org.apache.hadoop.io.compress.BZip2Codec") 

UPDATE 約1時間後、メモリ（72ギガバイト）は完全にいっぱいで、スワップ（72GB）もあります。私の場合、放送を使うことは解決策ではありません。

UPDATE 2 pickleで3つの変数をロードせずに、144GBの代わりに最大10GBのRAMを使用すると問題なく終了します。 （72GBのRAM + 72Gbのスワップ）

Answer 1

私の質問は：各エグゼクティブコアの各変数をsparkレプリケートしますか？

はい！

各（ローカル）変数のコピー数は、Pythonワーカーに割り当てるスレッドの数と同じです。あなたの問題については

---

、pickleを使用せずにcomp_regex、comp_dictとcomp_dict_legalをロードしてみてください。