相关疑难解决方法(0)

我有一个包含许多小.gz文件的文件夹(压缩的csv文本文件).我需要在我的Spark工作中阅读它们,但问题是我需要根据文件名中的信息进行一些处理.因此,我没有使用:

JavaRDD<<String>String> input = sc.textFile(...)

因为据我所知,我无法以这种方式访问​​文件名.相反,我用过:

JavaPairRDD<<String>String,String> files_and_content = sc.wholeTextFiles(...);

因为这样我得到了一对文件名和内容.但是,似乎这样,输入阅读器无法从gz文件中读取文本,而是读取二进制Gibberish.

所以,我想知道我是否可以将其设置为以某种方式读取文本,或者使用以下方式访问文件名 sc.textFile(...)

我的输入数据集大约是 150G。我正在设置

--conf spark.cores.max=100 
--conf spark.executor.instances=20 
--conf spark.executor.memory=8G 
--conf spark.executor.cores=5 
--conf spark.driver.memory=4G

但由于数据在执行者之间分布不均匀，我一直在得到

Container killed by YARN for exceeding memory limits. 9.0 GB of 9 GB physical memory used

这是我的问题：

1. Did I not set up enough memory in the first place? I think 20 * 8G > 150G, but it's hard to make perfect distribution, so some executors will suffer
2. I think about repartition the input dataFrame, so how can I determine how many partition to set? the …

我正在做一个实验，以了解s3和[EMR + Spark]哪种文件大小表现最佳

输入数据：不可压缩数据：文件中的随机字节总数据大小：20GB每个文件夹具有不同的输入文件大小：从2MB到4GB文件大小。

集群规格：1个主节点+ 4个节点：C3.8xls-驱动程序内存5G \-执行程序内存3G \-执行程序核心2 \-执行程序60 \

代码：

scala> def time[R](block: => R): R = {
          val t0 = System.nanoTime()
          val result = block    // call-by-name
         val t1 = System.nanoTime()
          println("Elapsed time: " + (t1 - t0) + "ns")
          result
      }
time: [R](block: => R)R

scala> val inputFiles = time{sc.textFile("s3://bucket/folder/2mb-10240files-20gb/*/*")};
scala> val outputFiles = time {inputFiles.saveAsTextFile("s3://bucket/folder-out/2mb-10240files-20gb/")};

观察=>

• 2MB-32MB：大多数时间都用于打开文件句柄[效率不高]
• 从64MB到1GB：Spark本身针对所有这些文件大小启动320个任务，它不再具有20GB数据的存储桶中的文件数量，例如512 mb文件具有40个文件来生成20gb数据，并且可以完成40个任务，但是而是有320个任务，每个任务处理64MB数据。
• 4GB文件大小：0字节输出[无法处理内存/数据甚至不可拆分???]

问题=>

• 强制处理输入大小的任何默认设置为64MB？
• 由于我正在使用的数据是随机字节并且已经压缩，因此如何进一步拆分此数据？如果可以拆分此数据，为什么无法拆分4gb目标文件大小的文件大小？
• 通过Spark上传后，为什么压缩文件的大小会增加？2MB的压缩输入文件在输出存储区中变为3.6 MB。