0
在我的HDFS中，我收集了大约350个csv文件。每个文件的大小从几个KB到250Mb不等。我需要将这些csv文件的值插入到名为RECORD的表中。插入时，我也需要引用一些其他表（PARAMETER和FRAME_RATE）。我有以下查询来完成此任务。将记录插入大量csv文件中的表格
-- create external table for the csv files in hdfs
CREATE EXTERNAL TABLE TEMP_CSV(
FRAME_RANK BIGINT,
FRATE BIGINT,
SOURCE STRING,
PARAM STRING,
RECORDEDVALUE STRING
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ';'
location '/user/bala/output'
TBLPROPERTIES ("skip.header.line.count"="2");
-- Now insert fresh values into T_RECORD
INSERT OVERWRITE TABLE RECORD
PARTITION(SESSION)
SELECT DISTINCT
TEMP_CSV.F_FRAME_RANK,
PARAMETER.K_ID,
FRAME_RATE.K_ID,
CAST(TEMP_CSV.RECORDEDVALUE as FLOAT),
split(reverse(split(reverse(TEMP_CSV.INPUT__FILE__NAME),"/")[0]), "[.]")[0] AS SESSION
FROM TEMP_CSV , PARAMETER, FRAME_RATE
WHERE PARAMETER.NAME = TEMP_CSV.PARAM AND FRAME_RATE.FRATE = TEMP_CSV.FRATE;
在我小的PoC的研究，我有大约50 CSV文件，该查询成功地填充记录到记录表中关于500秒与下面的配置
Hive-on-spark
spark standalon
6 nodes in the cluster
4 cores per node/16gb RAM
spark.executor.memory 2g
然而，当我处理350个文件，查询失败，执行程序中出现java堆空间错误。所以，我增加了executor.memory到4g。失败。我增加了executor.memory到6g。失败。最后，我增加了spark.executor.memory到12g。成功。但花了大约2小时30分钟。将spark.executor.memory增加到12g导致每个节点只有一个执行器，因此只有6个执行器。
当我的executor.memory是6G，这是日志在发生故障时，
******
******
2017-06-12 11:59:09,988 Stage-1_0: 101/101 Finished Stage-2_0: 12/12 Fini shed Stage-3_0: 0(+12,-2)/12
2017-06-12 11:59:12,997 Stage-1_0: 101/101 Finished Stage-2_0: 12/12 Finished Stage-3_0: 0(+12,-2)/12
2017-06-12 11:59:16,004 Stage-1_0: 101/101 Finished Stage-2_0: 12/12 Finished Stage-3_0: 0(+12,-2)/12
2017-06-12 11:59:19,012 Stage-1_0: 101/101 Finished Stage-2_0: 12/12 Finished Stage-3_0: 0(+12,-2)/12
*****
*****
在执行，这是错误日志
17/06/12 11:58:36 WARN NettyRpcEndpointRef: Error sending message [message = Heartbeat(5,[Lscala.Tuple2;@e65f7b8,BlockManagerId(5, bndligpu04, 54618))] in 1 attempts
org.apache.spark.rpc.RpcTimeoutException: Futures timed out after [50 seconds]. This timeout is controlled by spark.executor.heartbeatInterval
at org.apache.spark.rpc.RpcTimeout.org$apache$spark$rpc$RpcTimeout$$createRpcTimeoutException(RpcTimeout.scala:48)
at org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:63)
at org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:59)
at scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:33)
at org.apache.spark.rpc.RpcTimeout.awaitResult(RpcTimeout.scala:76)
at org.apache.spark.rpc.RpcEndpointRef.askWithRetry(RpcEndpointRef.scala:101)
at org.apache.spark.executor.Executor.org$apache$spark$executor$Executor$$reportHeartBeat(Executor.scala:476)
at org.apache.spark.executor.Executor$$anon$1$$anonfun$run$1.apply$mcV$sp(Executor.scala:505)
at org.apache.spark.executor.Executor$$anon$1$$anonfun$run$1.apply(Executor.scala:505)
at org.apache.spark.executor.Executor$$anon$1$$anonfun$run$1.apply(Executor.scala:505)
at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1801)
at org.apache.spark.executor.Executor$$anon$1.run(Executor.scala:505)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.util.concurrent.TimeoutException: Futures timed out after [50 seconds]
at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:219)
at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:223)
at scala.concurrent.Await$$anonfun$result$1.apply(package.scala:107)
at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:53)
at scala.concurrent.Await$.result(package.scala:107)
at org.apache.spark.rpc.RpcTimeout.awaitResult(RpcTimeout.scala:75)
... 14 more
17/06/12 11:58:36 ERROR Executor: Exception in task 0.0 in stage 3.0 (TID 115)
java.lang.OutOfMemoryError: Java heap space
at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
at org.apache.orc.impl.OutStream.getNewInputBuffer(OutStream.java:109)
at org.apache.orc.impl.OutStream.write(OutStream.java:130)
at org.apache.orc.impl.RunLengthIntegerWriterV2.writeDeltaValues(RunLengthIntegerWriterV2.java:238)
at org.apache.orc.impl.RunLengthIntegerWriterV2.writeValues(RunLengthIntegerWriterV2.java:186)
at org.apache.orc.impl.RunLengthIntegerWriterV2.write(RunLengthIntegerWriterV2.java:772)
at org.apache.orc.impl.WriterImpl$IntegerTreeWriter.writeBatch(WriterImpl.java:1039)
at org.apache.orc.impl.WriterImpl$StructTreeWriter.writeRootBatch(WriterImpl.java:1977)
at org.apache.orc.impl.WriterImpl.addRowBatch(WriterImpl.java:2759)
at org.apache.hadoop.hive.ql.io.orc.WriterImpl.flushInternalBatch(WriterImpl.java:277)
at org.apache.hadoop.hive.ql.io.orc.WriterImpl.addRow(WriterImpl.java:296)
at org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat$OrcRecordWriter.write(OrcOutputFormat.java:103)
at org.apache.hadoop.hive.ql.exec.FileSinkOperator.process(FileSinkOperator.java:743)
at org.apache.hadoop.hive.ql.exec.Operator.forward(Operator.java:837)
at org.apache.hadoop.hive.ql.exec.SelectOperator.process(SelectOperator.java:97)
at org.apache.hadoop.hive.ql.exec.spark.SparkReduceRecordHandler.processKeyValues(SparkReduceRecordHandler.java:309)
at org.apache.hadoop.hive.ql.exec.spark.SparkReduceRecordHandler.processRow(SparkReduceRecordHandler.java:267)
at org.apache.hadoop.hive.ql.exec.spark.HiveReduceFunctionResultList.processNextRecord(HiveReduceFunctionResultList.java:49)
at org.apache.hadoop.hive.ql.exec.spark.HiveReduceFunctionResultList.processNextRecord(HiveReduceFunctionResultList.java:28)
at org.apache.hadoop.hive.ql.exec.spark.HiveBaseFunctionResultList$ResultIterator.hasNext(HiveBaseFunctionResultList.java:95)
at scala.collection.convert.Wrappers$JIteratorWrapper.hasNext(Wrappers.scala:41)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at org.apache.spark.rdd.AsyncRDDActions$$anonfun$foreachAsync$1$$anonfun$apply$15.apply(AsyncRDDActions.scala:120)
at org.apache.spark.rdd.AsyncRDDActions$$anonfun$foreachAsync$1$$anonfun$apply$15.apply(AsyncRDDActions.scala:120)
at org.apache.spark.SparkContext$$anonfun$37.apply(SparkContext.scala:1992)
at org.apache.spark.SparkContext$$anonfun$37.apply(SparkContext.scala:1992)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
at org.apache.spark.scheduler.Task.run(Task.scala:89)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:227)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
17/06/12 11:58:36 ERROR SparkUncaughtExceptionHandler: Uncaught exception in thread Thread[Executor task launch worker-1,5,main]
java.lang.OutOfMemoryError: Java heap space
at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
at org.apache.orc.impl.OutStream.getNewInputBuffer(OutStream.java:109)
at org.apache.orc.impl.OutStream.write(OutStream.java:130)
at org.apache.orc.impl.RunLengthIntegerWriterV2.writeDeltaValues(RunLengthIntegerWriterV2.java:238)
我的问题是： -
- 是否有一个范围来优化查询？
- 任何其他可以应对这一挑战的火花/蜂巢配置？
- 有没有办法告诉Hive处理50个文件？
任何帮助/信息来解决这个问题将是有益的。还有一个信息，'SELECT'声明有效，我可以在我的色调浏览器中看到结果。当我尝试插入由'SELECT'收集的信息时，查询中断。
感谢您的建议，但我不知道如何--executor-cores = 2将是正确的修复。因为在一个核心中只有一个进程在6gb可用的情况下运行，它会失败。再运行一个进程与6gb共享应该再次失败。对？我可以尝试一下。 – Bala