スパークデータフレームランダム分割

Question

私は、列車、妥当性検査、およびテストに比率0.60,0.20,0.20で分割したいスパークデータフレームを持っています。

Iは同一のため、次のコードを使用する：

def data_split(x): 
    global data_map_var 
    d_map = data_map_var.value 
    data_row = x.asDict() 
    import random 
    rand = random.uniform(0.0,1.0) 
    ret_list =() 
    if rand <= 0.6: 
     ret_list = (data_row['TRANS'] , d_map[data_row['ITEM']] , data_row['Ratings'] , 'train') 
    elif rand <=0.8: 
     ret_list = (data_row['TRANS'] , d_map[data_row['ITEM']] , data_row['Ratings'] , 'test') 
    else: 
     ret_list = (data_row['TRANS'] , d_map[data_row['ITEM']] , data_row['Ratings'] , 'validation') 
    return ret_list 
​ 
​ 
split_sdf = ratings_sdf.map(data_split) 
train_sdf = split_sdf.filter(lambda x : x[-1] == 'train').map(lambda x :(x[0],x[1],x[2])) 
test_sdf = split_sdf.filter(lambda x : x[-1] == 'test').map(lambda x :(x[0],x[1],x[2])) 
validation_sdf = split_sdf.filter(lambda x : x[-1] == 'validation').map(lambda x :(x[0],x[1],x[2])) 
​ 
print "Total Records in Original Ratings RDD is {}".format(split_sdf.count()) 
​ 
print "Total Records in training data RDD is {}".format(train_sdf.count()) 
​ 
print "Total Records in validation data RDD is {}".format(validation_sdf.count()) 
​ 
print "Total Records in test data RDD is {}".format(test_sdf.count()) 
​ 
​ 
#help(ratings_sdf) 
Total Records in Original Ratings RDD is 300001 
Total Records in training data RDD is 180321 
Total Records in validation data RDD is 59763 
Total Records in test data RDD is 59837 

私の元のデータフレームがIに分割を行いマッパー関数を渡すために使用ratings_sdfあります。

列車の合計を確認すると、検証とテストはスプリット（元の評価）の合計とはなりません。そして、これらの数字はコードの実行ごとに変わります。

残りのレコードはどこにありますか？また、合計が等しくない理由は何ですか？

Answer 1

TL; DRあなたはDataFrame使用randomSplit methodを分割したい場合：

ratings_sdf.randomSplit([0.6, 0.2, 0.2]) 

あなたのコードは、複数のレベルでちょうど間違っているが、それは修復不可能壊れ作る二つの基本的な問題がある：

は、

• スパーク変換は任意の回数評価することができ、使用する関数は明確に透明で副作用がなくてはなりません。コードでsplit_sdfが複数回評価され、ステートフルなRNG data_splitが使用されるため、結果が異なる場合があります。

  これは、各子が親RDDのさまざまな状態を表示する動作を示します。

• あなたはRNGを適切に初期化していないため、得られるランダムな値は独立していません。