如何在 spark rdd 中读取 zip 文件中的分隔文件

Question

我有一个 HDFS 位置，并且该位置中有一个 zip 文件

HDFS 位置 /development/staging/b8baf3f4-abce-11eb-8592-0242ac110032/records.zip

 scala> val loc = "/development/staging/b8baf3f4-abce-11eb-8592-0242ac110032/"
 loc: String = "/development/staging/b8baf3f4-abce-11eb-8592-0242ac110032/"

 scala> val rdd = sc.textFile(loc)
 rdd: org.apache.spark.rdd.RDD[String] = /development/staging/b8baf3f4-abce-11eb-8592-0242ac110032/ MapPartitionsRDD[1] at textFile at <console>:26

  scala> rdd.take(2)
  res0: Array[String] = Array(PK????????]R�R��*�????�??? ???2972120.dat�S�r�0? 
   ��*�0����?t?�]T�Ж??����
 `�6ط�kU;P�M�� rSO�;G��p��?��?�Z1^3@�^�� ��F��ٕb�?~,ٖ 
 �u6�D��'�@�??��L*�Gp?�kcL�7!r�p1�1e�� a*.{? 
   �.;��������s�(�)�, ?�=�9U<"*!?5��?;�?�?�مd{h} 
  ��gG���� �?�Z)

但它产生的输出不同

你能帮助我如何使用 spark RDD 读取 zip 文件中的文件吗？我的 zip 文件中只有一个文件

Answer 1

你在寻找这样的东西吗：

import java.io.{ IOException, FileOutputStream, FileInputStream, File }
import java.util.zip.{ ZipEntry, ZipInputStream }
import org.apache.spark.sql.SQLContext
import org.apache.spark.SparkContext

//Unzip the file and copy the internal contents outside in new location
object Unzip extends App {

  val INPUT_ZIP_FILE: String = "src/resources/my-zip.zip";
val OUTPUT_FOLDER: String = "src/resources/my-zip";

def unZipIt(zipFile: String, outputFolder: String): Unit = {

  val buffer = new Array[Byte](1024)

  try {

    //output directory
    val folder = new File(OUTPUT_FOLDER);
    if (!folder.exists()) {
      folder.mkdir();
    }

    //zip file content
    val zis: ZipInputStream = new ZipInputStream(new FileInputStream(zipFile));
    //get the zipped file list entry
    var ze: ZipEntry = zis.getNextEntry();

    while (ze != null) {

      val fileName = ze.getName();
      val newFile = new File(outputFolder + File.separator + fileName);

      System.out.println("file unzip : " + newFile.getAbsoluteFile());

      //create folders
      new File(newFile.getParent()).mkdirs();

      val fos = new FileOutputStream(newFile);

      var len: Int = zis.read(buffer);

      while (len > 0) {

        fos.write(buffer, 0, len)
        len = zis.read(buffer)
      }

      fos.close()
      ze = zis.getNextEntry()
    }

    zis.closeEntry()
    zis.close()

  } catch {
    case e: IOException => println("exception caught: " + e.getMessage)
  }

}

Unzip.unZipIt(INPUT_ZIP_FILE, OUTPUT_FOLDER)
val sac = new SparkContext("local[*]", " first Program");
val sqlc = new SQLContext(sac);
val rdd = sac.textFile("src/resources/my-zip/sample.txt")
rdd.take(1).foreach(println)
/*val rddFromFile = sqlc.sparkContext.textFile("src/resources/my-zip/sample.txt")
println(rddFromFile.getClass)

println("##Get data Using collect")
rddFromFile.collect().foreach(f=>{
  println(f)
})*/

}

不确定这是否能实现您想要做的，但可能会有所帮助！