我正在加入 1100 万条记录。我在 EMR Cluster Spark 2.2.1 中与 5 个工作人员一起运行
运行作业时出现以下错误：
executor 3): java.lang.IllegalArgumentException: Cannot allocate a page with more than 17179869176 bytes
at org.apache.spark.memory.TaskMemoryManager.allocatePage(TaskMemoryManager.java:277)
at org.apache.spark.memory.MemoryConsumer.allocateArray(MemoryConsumer.java:90)
at org.apache.spark.shuffle.sort.ShuffleExternalSorter.growPointerArrayIfNecessary(ShuffleExternalSorter.java:328)
at org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:379)
at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:246)
at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:167)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
at org.apache.spark.scheduler.Task.run(Task.scala:108)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:338)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
我无法理解可能的原因。请帮我设置什么参数。
目前我正在运行以下参数： --num-executors 5 --conf spark.eventLog.enabled=true --executor-memory 70g --driver-memory 30g --executor-cores 16 --conf spark.shuffle.memoryFraction=0.5
我是Apache Spark(版本1.4.1)的新手.我写了一个小代码来读取文本文件并将其数据存储在Rdd中.
有没有办法在rdd中获取数据大小.
这是我的代码:
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.util.SizeEstimator
import org.apache.spark.sql.Row
object RddSize {
def main(args: Array[String]) {
val sc = new SparkContext("local", "data size")
val FILE_LOCATION = "src/main/resources/employees.csv"
val peopleRdd = sc.textFile(FILE_LOCATION)
val newRdd = peopleRdd.filter(str => str.contains(",M,"))
//Here I want to find whats the size remaining data
}
}
我希望在过滤转换(peopleRdd)之前和之后(newRdd)获取数据大小.
我试图在Apache spark sql上运行查询.第一个查询工作正常,但第二个查询也删除空值.
代码:
def main(args: Array[String]) {
val sc = new SparkContext("local[*]", "Spark")
val sqlContext = new SQLContext(sc)
val pageViewsDF = getDataframe(sc, sqlContext)
println("RUNNING SQL QUERIES ")
sqlContext.sql("select name , count(*) from pageviews_by_second group by name").show(10)
sqlContext.sql("select name , count(*) from pageviews_by_second where name not in (\"Rose\") group by name").show(10)
}
def getDataframe(sc: SparkContext, sqlContext: SQLContext): DataFrame = {
Logger.getLogger("org").setLevel(Level.OFF);
Logger.getLogger("akka").setLevel(Level.OFF);
val dataArray = List(List("David", null),
List("David", null),
List("Charlie", "23"),
List("Rose", null),
List("Ben", null),
List("Harry", "43"),
List(null, "25"),
List(null, …wordsDF = sqlContext.createDataFrame([('cat',), ('elephant',), ('rat',), ('rat',), ('cat', )], ['word'])
这是一种从python中的元组列表创建数据框的方法。如何在Scala中做到这一点？我是Scala的新手，在解决它时遇到了问题。
任何帮助将不胜感激！