डेटाफ्रेम को संपीड़ित (gzipped) CSV के रूप में कैसे सहेजना है?

Question

मैं स्पार्क 1.6.0 और स्कैला का उपयोग करता हूं।

मैं डेटाफ्रेम को संकुचित सीएसवी प्रारूप के रूप में सहेजना चाहता हूं।

//set the conf to the codec I want 
sc.getConf.set("spark.hadoop.mapred.output.compress", "true") 
sc.getConf.set("spark.hadoop.mapred.output.compression.codec", "true") 
sc.getConf.set("spark.hadoop.mapred.output.compression.codec", "org.apache.hadoop.io.compress.GzipCodec") 
sc.getConf.set("spark.hadoop.mapred.output.compression.type", "BLOCK") 

df.write 
    .format("com.databricks.spark.csv") 
    .save(my_directory) 

उत्पादन gz स्वरूप में नहीं है:

यहाँ मैं अब तक है (मान मैं पहले से ही SparkContext रूप df और sc है)।

Answer 1

चिंगारी से सीएसवी GitHub पर: https://github.com/databricks/spark-csv

एक पढ़ सकते हैं:

codec: संपीड़न कोडेक जब दाखिल करने के लिए बचत का उपयोग करें। Org.apache.hadoop.io.compress.CompressionCodec या केस-असंवेदनशील शॉर्ट नामों में से एक (bzip2, gzip, lz4, और snappy) को लागू करने वाले वर्ग का पूर्णतः योग्य नाम होना चाहिए। कोडेक निर्दिष्ट नहीं होने पर कोई संपीड़न के लिए डिफ़ॉल्ट।

आपके मामले में, यह काम करना चाहिए: df.write.format("com.databricks.spark.csv").codec("gzip")\ .save('my_directory/my_file.gzip')

Answer 2

इस कोड स्पार्क 2.1, जहां .codec उपलब्ध नहीं है के लिए काम करता है।

df.write 
    .format("com.databricks.spark.csv") 
    .option("codec", "org.apache.hadoop.io.compress.GzipCodec") 
    .save(my_directory) 

स्पार्क 2.2 के लिए, आप df.write.csv(...,codec="gzip") विकल्प यहाँ वर्णित का उपयोग कर सकते हैं: https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=codec

Answer 3

स्पार्क के साथ

2.0+, इस बन गया है एक सरल सा:

df.write.csv("path", compression="gzip") 

आप की जरूरत नहीं है बाहरी डाटाबेस सीएसवी पैकेज अब और।

csv() लेखक कई आसान विकल्पों का समर्थन करता है। उदाहरण के लिए:

• sep: विभाजक चरित्र सेट करने के लिए।
• quote: मूल्यों को उद्धृत करने के लिए और कैसे करें।
• header: हेडर लाइन को शामिल करना है या नहीं।

वहाँ भी gzip करने के लिए अन्य संपीड़न कोडेक आप उपयोग कर सकते हैं, इसके अलावा में की एक संख्या हैं:

• bzip2
• lz4
• snappy
• deflate

csv() लेखक के लिए पूर्ण स्पार्क डॉक्स यहां हैं: Python/Scala

Answer 4

सीएसवी हेडर के साथ फाइल लिखने और .csv.gzip को

DF.coalesce(1).write.format("com.databricks.spark.csv").mode("overwrite") 
.option("header","true") 
.option("codec",org.apache.hadoop.io.compress.GzipCodec").save(tempLocationFileName) 

copyRename(tempLocationFileName, finalLocationFileName) 

def copyRename(srcPath: String, dstPath: String): Unit = { 
    val hadoopConfig = new Configuration() 
    val hdfs = FileSystem.get(hadoopConfig) 
    FileUtil.copyMerge(hdfs, new Path(srcPath), hdfs, new Path(dstPath), true, hadoopConfig, null) 
    // the "true" setting deletes the source files once they are merged into the new output 
} 

ऐसा न करने पर भाग-000 फ़ाइल का नाम बदलने के लिए हेडर की आवश्यकता है और फिर इसे गलत पर सेट करें और आपको कोलेसे को भी करने की आवश्यकता नहीं होगी। यह भी लिखना तेज़ होगा।