PySparkでは、値を比較して別のRDDを返す関数にRDDを送信するにはどうすればよいですか？

Question

は、我々は次のサンプルデータを持っていると言う：私は、サンプルファイルの各行、例えば1、2、3、0、3、1の最後の値をチェックしたい

1,John,Martinez,North Lauderdale,20160101,1 
2,John,Martinez,Plantation,20170101,2 
3,John,Martinez,North Lauderdale,20161022,1 
4,John,Martinez,Pembroke Pines,20181231,0 
5,John,Martinez,Plantation,20190101,3 
6,John,Martinez,Plantation,20200101,1 
7,John,Martinez,Plantation,20210101,9 

を、9

def func(input): 
    if str(input[5]) is "1": 
     rdd_trdln = input.map(lambda line: (line, "A")) 
    else: 
     rdd_trdln = input.map(lambda line: (line, "O")) 
     return rdd_trdln 
input = sc.textFile("file.txt").map(lambda line: line.split('\t')) 
return_FirstFunc = input.map(firstFunc) 

私は取得していますエラー：

AttributeError: 'list' object has no attribute 'map'

Answer 1

Spark RDD.map()と通常のPython map()の機能には違いがあります。

sc.textFile("file.txt").map(lambda line: line.split('\t'))があるときは、PythonリストのRDDを作成しました。したがって、input.map(func)に電話するときは、funcはのリストを受け入れる必要があります.RDDは受け付けません。

したがって、input.mapはあなたのエラー...

'list' object has no attribute 'map'

があるそしてそれはPythonのエラーではなく、スパーク・エラーです。

あなただけのリストの中に文字を追加したい場合は、あなたのコードは

def func(input): 
    if input[5] == "1": 
     input.append("A") 
    else: 
     input.append("O") 
    return input 

それとも、もっとニシキヘビ

def func(input): 
    input.append("A" if input[5] == "1" else "O") 
    return input 

---

それともあなたが取るためにあなたの関数を定義することができますでしょう文字列としての行全体がそこに分割されます。
リストのRDDを使用すると、作業が面倒で時々覚えにくくなります。あなたがここにそう

for line in converted_lines.collect(): 
    print(line) 

Answer 2

純粋なPythonの側からは、あなたが標準listの上にマップする場合、あなたはbuilt-in map functionを使用することができます。 python3は、Python 2（リスト）に異なる結果タイプ（map反復子）を生成

input = map(lambda line: line.split('\t'), sc.textFile("file.txt")) 

注意。