PySpark

Question

にカスタム関数を適用し、次のスニペットはPySpark RDDオブジェクトに単純な関数を適用しようとすると、外部モジュールを使用する：

import pyspark 
conf = pyspark.SparkConf() 
conf.set('spark.dynamicAllocation.minExecutors', 5) 
sc = SparkContext(appName="tmp", conf=conf) 
sc.setLogLevel('WARN') 

fn = 'my_csv_file' 
rdd = sc.textFile(fn) 
rdd = rdd.map(lambda line: line.split(",")) 
header = rdd.first() 
rdd = rdd.filter(lambda line:line != header) 
def parse_line(line): 
    ret = pyspark.Row(**{h:line[i] for (i, h) in enumerate(header)}) 
    return ret 
rows = rdd.map(lambda line: parse_line(line)) 
sdf = rows.toDF() 

私はpython my_snippet.pyでプログラムを起動すると、それはそれを文句で失敗します。

def parse_line(line): 
    ret = h:line[i] for (i, h) in enumerate(header) 
    ret['dir'] = dir() 
    return ret 

：

File "<ipython-input-27-8e46d56b2984>", line 6, in <lambda> 
File "<ipython-input-27-8e46d56b2984>", line 3, in parse_line 
NameError: global name 'pyspark' is not defined 

は、私は次のようにparse_line機能を置き換えます

データフレームが作成され、dirの列には、 内のネームスペースに、lineとretという2つのオブジェクトしか含まれていないことが示されています。関数の一部として他のモジュールとオブジェクトを持つにはどうすればいいですか？ pysparkだけでなく他のものもあります。

EDITこのpysparkは、プログラム内で使用できます。関数がmapによって呼び出された場合（そしてfilter、reduceなどと仮定しています）、インポートされたモジュールは表示されません。

Answer 1

1）元の質問への回答： 問題の原因は、あなたは使用してコードを実行する必要がありmy_snippet.py のpython火花を提出実行されているように思えmy_snippet.py

2）次の行が存在しない、私のipythonのノート型パーソナル設定で ：

は

import pyspark 
conf = pyspark.SparkConf() 
conf.set('spark.dynamicAllocation.minExecutors', 5) 
sc = SparkContext(appName="tmp", conf=conf) 

「SC」の範囲外で定義されているノートブックの質問に答えるためにipython私のプログラム

3）numpyのを使用するためにnumpyの（またはインストールする必要が他のモジュール） に関する質問への答えは、あなたが）のapt-getまたはピップまたはソースからのインストールを使用して（numpyのをインストールする必要が上のすべてのクラスタ内のノード。