spark-1.5.1 在 java 代码中使用 HiveContext 为 hive 1.2.0 抛出内存不足错误答案

【问题标题】：spark-1.5.1 throwing out of memory error for hive 1.2.0 using HiveContext in java codespark-1.5.1 在 java 代码中使用 HiveContext 为 hive 1.2.0 抛出内存不足错误
【发布时间】：2016-03-16 06:17:57
【问题描述】：

我有一个用于 HADOOP 2.6 的 spark-1.5.1，在我的本地计算机上以独立模式运行。我正在尝试从示例 java 应用程序运行配置单元查询，将 spark.master 指向在我的本地计算机上运行的 (spark://impetus-i0248u:7077) spark master。这是一段java代码：

 SparkConf sparkconf = new SparkConf().set("spark.master", "spark://impetus-i0248u:7077").set("spark.app.name", "sparkhivesqltest")
        .set("spark.cores.max", "2").set("spark.executor.memory", "2g").set("worker_max_heapsize","2g").set("spark.driver.memory", "2g");

 SparkContext sc = new SparkContext(sparkconf);

HiveContext sqlContext = new HiveContext(sc);
DataFrame jdbcDF = sqlContext.sql("select * from bm.rutest");

List<Row> employeeFullNameRows = jdbcDF.collectAsList();

HiveContext 正在正确初始化，因为它能够与 hive 元存储建立连接。我在jdbcDF.collectAsList(); 遇到了例外情况@

这是 spark 尝试提交作业时出现的错误：

提交 15/12/10 20:00:42 INFO DAGScheduler: 提交 2 个缺失 ResultStage 0 (MapPartitionsRDD[3] at collectAsList at HiveJdbcTest.java:30) 2010 年 15 月 12 日 20:00:42 信息 TaskSchedulerImpl：添加任务集 0.0 和 2 个任务 2010 年 15 月 12 日 20:00:42 信息 TaskSetManager：在阶段 0.0 中启动任务 0.0（TID 0、172.26.52.54、ANY、2181 字节） 15/12/10 20:00:42 INFO TaskSetManager：在阶段 0.0 中启动任务 1.0 (TID 1, 172.26.52.54, ANY, 2181 字节)

异常：java.lang.OutOfMemoryError 从线程中的 UncaughtExceptionHandler “sparkDriver-akka.remote.default-remote-dispatcher-5”中的异常线程“shuffle-server-1”异常：java.lang.OutOfMemoryError 从线程“shuffle-server-1”中的 UncaughtExceptionHandler 抛出线程“threadDeathWatcher-2-1”中的异常异常： java.lang.OutOfMemoryError 从 UncaughtExceptionHandler 中抛出线程“threadDeathWatcher-2-1”

异常：java.lang.OutOfMemoryError 从线程中的 UncaughtExceptionHandler “sparkDriver-akka.remote.default-remote-dispatcher-6”中的异常线程“qtp1003369013-56”异常：java.lang.OutOfMemoryError 从线程“qtp1003369013-56”中的 UncaughtExceptionHandler 抛出

异常：java.lang.OutOfMemoryError 从线程中的 UncaughtExceptionHandler “sparkDriver-akka.remote.default-remote-dispatcher-21”

异常：java.lang.OutOfMemoryError 从线程中的 UncaughtExceptionHandler “sparkDriver-akka.actor.default-dispatcher-17”

异常：java.lang.OutOfMemoryError 从线程中的 UncaughtExceptionHandler “sparkDriver-akka.remote.default-remote-dispatcher-23”中的异常线程“shuffle-server-2”异常：java.lang.OutOfMemoryError 从线程“shuffle-server-2”中的 UncaughtExceptionHandler 抛出

异常：java.lang.OutOfMemoryError 从线程中的 UncaughtExceptionHandler “sparkDriver-akka.actor.default-dispatcher-2”

下面是spark-env.sh中添加的配置

SPARK_EXECUTOR_CORES=2
SPARK_EXECUTOR_MEMORY=3G
SPARK_WORKER_CORES=2
SPARK_WORKER_MEMORY=2G
SPARK_EXECUTOR_INSTANCES=2
SPARK_WORKER_INSTANCES=1

如果我将 spark.master 设置为本地 [*]，它可以正常工作，但是当我将它指向在我的机器上运行的 master 时，我得到了上面提到的异常。如果我尝试使用相同的配置连接到 mysql db，它可以正常工作。

PS：表格只有一行。

请帮忙..！

【问题讨论】：

collect() 方法收集所有工作人员的所有数据并将其存储在驱动程序上，使用驱动程序内存。如果您在大型数据集上运行，那么您将获得 OOM 是合乎逻辑的。考虑天气collect() 是您真正寻找的行动。
我想要给定查询的所有记录集，在这种情况下，我的表只有一行。
为什么要将它返回给您的驱动程序？什么原因？
能够使用这段代码而不是使用collectAsList()来解决这个OOM错误。迭代器 iter = jdbcDF.javaRDD().toLocalIterator(); while (iter.hasNext()){ Row row = iter.next(); } 谢谢 :)
但我还是没明白，为什么 collectAsList() 会为单个记录抛出 OOM 错误。即使我配置了 2GB 的驱动程序内存，仍然面临同样的错误。你能详细说明一下吗？我是 Spark 的新手..

标签： java hadoop apache-spark hive apache-spark-sql

【解决方案1】：

以下是对您问题中概念的解释：-

local[*] = 执行是多线程的，不是分布式的。适合在单机上测试作业时的开发。它适用于您的情况，因为数据没有被洗牌或从 Executors 移动到驱动程序......所有都在一个单一的 JVM 和本地。
collectAsList - 此方法将从驱动程序节点上的 Executor 收集所有数据，这会导致 Shuffle 和 Shuffle 是一个内存密集型过程，因为它需要序列化、网络和磁盘 IO。
javaRDD().toLocalIterator() = 产生与 collect() 相同的结果，但按顺序在每个分区上工作，不涉及洗牌。请注意，我们在 RDD 中仅使用这种分区顺序，并且分区中项目的顺序是明确定义的。

所以考虑到上述情况，当您使用本地框时，很有可能 local(*) 或 collectAsList() 可能不会给出任何 OOM，但 collect() 可能会产生内存异常。

【讨论】：