Speichern Sie die Datei in HDFS von einem Paar RDD

Question

Unten ist das Python-Skript, das ich zum Schreiben in HDFS verwende. RDD ist ein Paar RDD. Das Skript funktioniert gut, aber es erstellt einen Eintrag als Tupel in HDFS.Ist ist es möglich, das Tupel zu entfernen und nur durch Kommas getrennte Einträge in HDFS zu erstellen.

import sys 
from pyspark import SparkContext 

if len(sys.argv) < 2: 
    print 'Insufficient arguments' 
    sys.exit() 

sc = SparkContext() 
initialrdd1 = sc.textFile(sys.argv[1]) 
finalRDD1 = initialrdd1.map(lambda x:x.split(',')).map(lambda x :(x[1],x[0])).sortByKey() 
print finalRDD1.getNumPartitions() 
finalRDD1.saveAsTextFile('/export_dir/result3/') 

Datei in HDFS Speicherung ist in folgendem Format

(u'Alpha', u'E03') 
(u'Beta', u'E02') 
(u'Gamma', u'E05') 
(u'Delta', u'E09') 

Answer 1

Warum nicht zuerst die Tupel zu bespannen Karte und speichern Sie es dann -

finalRDD1.map(lambda x: ','.join(str(s) for s in x)).saveAsTextFile('/export_dir/result3/') 

Answer 2

finalRDD1 = initialrdd1.map(lambda x:x.split(',')).map(lambda x :(x[1],x[0])).sortByKey() 

Code verstehen. In Ihrer anfänglichen RDD ordnen Sie jeden Eintrag einem Tupel zu. Karte (lambda x: (x [1], x [0]))

finalRDD1.saveAsTextFile('/export_dir/result3/') 

Nach dem sortByKey Betrieb direkt die RDD als Textdatei zu speichern fortfahren.

Um Einträge als CSV-Datei zu speichern, müssen Sie es wie so explizit angeben -

def csv_format(data): 
    return ','.join(str(d) for d in data) 

# Rest of the code ... 

finalRDD1.map(csv_format).saveAsTextFile('/export_dir/result3/') 

Answer 3

ich ähnliches Problem hatte. Das Problem mit

map(lambda x: ','.join(str(s) for s in x)).saveAsTextFile(....) 

ist, dass es als eine Zeichenfolge ‚nachziehen‘ sparen und das wird das Komma verstecken und könnte ein Problem werden, wenn Sie für die Analyse wie das Laden als Pandas df verwenden mögen. So werden Sie String so aussehen

[ 'Alpha, E03', 'Beta, E02',....] 

einfache Lösung wäre, eine andere Karte geschlitzten Tüllen vor saveAsTextFile()

.map(lambda x: x.split(',')).saveAsTextFile(....) 

So endgültige Code wie dieser

finalRDD1.map(csv_format).map(lambda x: ','.join(str(s) for s in x)).map(lambda x: x.split(',')).saveAsTextFile('/export_dir/result3/') 

aussehen wird nun Ihre csv wird so aussehen

[ 'Alpha', 'E03'] 
['Beta', 'E02'] 
..... 
.....