Remember that spark foreachpartition once led to oom

Problem description

spark streaming Program line reporting error
The log is as follows ：

org.apache.spark.SparkException: Job aborted due to stage failure: Task serialization failed: java.lang.OutOfMemoryError: GC overhead limit exceeded
java.lang.OutOfMemoryError: GC overhead limit exceeded
	at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
	at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
	at org.apache.spark.broadcast.TorrentBroadcast$.$anonfun$blockifyObject$1(TorrentBroadcast.scala:306)
	at org.apache.spark.broadcast.TorrentBroadcast$.$anonfun$blockifyObject$1$adapted(TorrentBroadcast.scala:306)
	at org.apache.spark.broadcast.TorrentBroadcast$$$Lambda$2411/66155661.apply(Unknown Source)
	at org.apache.spark.util.io.ChunkedByteBufferOutputStream.toChunkedByteBuffer(ChunkedByteBufferOutputStream.scala:114)
	at org.apache.spark.broadcast.TorrentBroadcast$.blockifyObject(TorrentBroadcast.scala:315)
	at org.apache.spark.broadcast.TorrentBroadcast.writeBlocks(TorrentBroadcast.scala:137)
	at org.apache.spark.broadcast.TorrentBroadcast.<init>(TorrentBroadcast.scala:91)
	at org.apache.spark.broadcast.TorrentBroadcastFactory.newBroadcast(TorrentBroadcastFactory.scala:35)
	at org.apache.spark.broadcast.BroadcastManager.newBroadcast(BroadcastManager.scala:77)
	at org.apache.spark.SparkContext.broadcast(SparkContext.scala:1479)
	at org.apache.spark.scheduler.DAGScheduler.submitMissingTasks(DAGScheduler.scala:1223)
	at org.apache.spark.scheduler.DAGScheduler.submitStage(DAGScheduler.scala:1118)
	at org.apache.spark.scheduler.DAGScheduler.handleJobSubmitted(DAGScheduler.scala:1061)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2196)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2188)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2177)
	at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)

---

Cause analysis ：

According to the log information Locate in the code dataFrame.foreachPartition

1.foreachPartition Introduce

2. It was used foreachPartition After operator , Where are the benefits ？

1、 For what we wrote function function , Just call it once , One at a time partition All data

2、 Mainly create or obtain a database connection

3、 Just send it to the database once SQL Statement and multiple sets of parameters

In the actual production environment , Clear and uniform , Is to use foreachPartition operation ; But there's a problem , Follow mapPartitions Same operation , If one partition The number of is really very large , For example, it's really 100 ten thousand , That's basically unreliable .

Come in at once , It's very likely that OOM, Memory overflow problem .

---

Solution ：

1. Program plus memory