apache-spark – 经过长时间正常运行后,Spark状态流媒体作业在检查点S3时挂起

2023年12月4日 241次阅读

我最近一直在压力测试我们的Spark Streaming应用程序.压力测试摄取大约20,000条消息/秒,消息大小在200字节之间变化 – 1K到Kafka,其中Spark Streaming每4秒读取一次批次.

我们的Spark集群使用独立集群管理器在1.6.1版本上运行,我们的代码使用Scala 2.10.6.

运行大约15-20小时后,其中一个启动检查点的执行程序(以40秒的间隔完成)卡在下面的堆栈跟踪中,并且永远不会完成：

java.net.SocketInputStream.socketRead0(Native Method)
java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
java.net.SocketInputStream.read(SocketInputStream.java:170)
java.net.SocketInputStream.read(SocketInputStream.java:141)
sun.security.ssl.InputRecord.readFully(InputRecord.java:465)
sun.security.ssl.InputRecord.readV3Record(InputRecord.java:593)
sun.security.ssl.InputRecord.read(InputRecord.java:532)
sun.security.ssl.SSLSocketImpl.readRecord(SSLSocketImpl.java:973)
sun.security.ssl.SSLSocketImpl.performInitialHandshake(SSLSocketImpl.java:1375)
sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:1403)
sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:1387)
org.apache.http.conn.ssl.SSLSocketFactory.connectSocket(SSLSocketFactory.java:533)
org.apache.http.conn.ssl.SSLSocketFactory.connectSocket(SSLSocketFactory.java:401)
org.apache.http.impl.conn.DefaultClientConnectionOperator.openConnection(DefaultClientConnectionOperator.java:177)
org.apache.http.impl.conn.AbstractPoolEntry.open(AbstractPoolEntry.java:144)
org.apache.http.impl.conn.AbstractPooledConnAdapter.open(AbstractPooledConnAdapter.java:131)
org.apache.http.impl.client.DefaultRequestDirector.tryConnect(DefaultRequestDirector.java:610)
org.apache.http.impl.client.DefaultRequestDirector.execute(DefaultRequestDirector.java:445)
org.apache.http.impl.client.AbstractHttpClient.doExecute(AbstractHttpClient.java:863)
org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:82)
org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:57)
org.jets3t.service.impl.rest.httpclient.RestStorageService.performRequest(RestStorageService.java:326)
org.jets3t.service.impl.rest.httpclient.RestStorageService.performRequest(RestStorageService.java:277)
org.jets3t.service.impl.rest.httpclient.RestStorageService.performRestHead(RestStorageService.java:1038)
org.jets3t.service.impl.rest.httpclient.RestStorageService.getObjectImpl(RestStorageService.java:2250)
org.jets3t.service.impl.rest.httpclient.RestStorageService.getObjectDetailsImpl(RestStorageService.java:2179)
org.jets3t.service.StorageService.getObjectDetails(StorageService.java:1120)
org.jets3t.service.StorageService.getObjectDetails(StorageService.java:575)
org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.retrieveMetadata(Jets3tNativeFileSystemStore.java:174)
sun.reflect.GeneratedMethodAccessor32.invoke(Unknown Source)
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
java.lang.reflect.Method.invoke(Method.java:497)
org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:187)
org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102)
org.apache.hadoop.fs.s3native.$Proxy18.retrieveMetadata(Unknown
Source)
org.apache.hadoop.fs.s3native.NativeS3FileSystem.getFileStatus(NativeS3FileSystem.java:472)
org.apache.hadoop.fs.FileSystem.exists(FileSystem.java:1424)
org.apache.spark.rdd.ReliableCheckpointRDD$.writePartitionToCheckpointFile(ReliableCheckpointRDD.scala:168)
org.apache.spark.rdd.ReliableCheckpointRDD$$anonfun$writeRDDToCheckpointDirectory$1.apply(ReliableCheckpointRDD.scala:136)
org.apache.spark.rdd.ReliableCheckpointRDD$$anonfun$writeRDDToCheckpointDirectory$1.apply(ReliableCheckpointRDD.scala:136)
org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
org.apache.spark.scheduler.Task.run(Task.scala:89)
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
java.lang.Thread.run(Thread.java:745)

卡住了,火花驱动器拒绝继续处理传入的批次,并创建了大量的排队批量积压,在释放“卡住”的任务之前无法处理.

此外,查看streaming-job-executor-0下的驱动程序线程转储清楚地表明它正在等待此任务完成：

java.lang.Object.wait(Native Method)
java.lang.Object.wait(Object.java:502)
org.apache.spark.scheduler.JobWaiter.awaitResult(JobWaiter.scala:73)
org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:612)
org.apache.spark.SparkContext.runJob(SparkContext.scala:1832)
org.apache.spark.SparkContext.runJob(SparkContext.scala:1845)
org.apache.spark.SparkContext.runJob(SparkContext.scala:1922)
org.apache.spark.rdd.ReliableCheckpointRDD$.writeRDDToCheckpointDirectory(ReliableCheckpointRDD.scala:135)
org.apache.spark.rdd.ReliableRDDCheckpointData.doCheckpoint(ReliableRDDCheckpointData.scala:58)
org.apache.spark.rdd.RDDCheckpointData.checkpoint(RDDCheckpointData.scala:74)
org.apache.spark.rdd.RDD$$anonfun$doCheckpoint$1.apply$mcV$sp(RDD.scala:1682)
org.apache.spark.rdd.RDD$$anonfun$doCheckpoint$1.apply(RDD.scala:1679)
org.apache.spark.rdd.RDD$$anonfun$doCheckpoint$1.apply(RDD.scala:1679)
org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
org.apache.spark.rdd.RDD.doCheckpoint(RDD.scala:1678)
org.apache.spark.rdd.RDD$$anonfun$doCheckpoint$1$$anonfun$apply$mcV$sp$1.apply(RDD.scala:1684)
org.apache.spark.rdd.RDD$$anonfun$doCheckpoint$1$$anonfun$apply$mcV$sp$1.apply(RDD.scala:1684)
scala.collection.immutable.List.foreach(List.scala:318)

有谁遇到过这样的问题？

最佳答案 由于org.jets3t使用的HttpClient库中的错误导致套接字挂起,其中SSL握手不使用指定的超时.您可以找到问题详情
here.

此错误在v4.5.1以下的HttpClient版本中重现,并在其中修复.不幸的是,Spark 1.6.x使用的是v4.3.2,它没有提供的修复程序.

到目前为止,我有三种可能的解决方法：

>通过spark.speculation配置设置使用Spark的推测机制.这有助于悬挂的边缘情况,因为它很少再现并且在负载下.请注意,这可能会导致流媒体作业开始时出现一些误报,其中火花并不能很好地了解您的中位数任务的运行时间,但绝对不会导致明显滞后.

文件说：

If set to “true”, performs speculative execution of tasks. This means
if one or more tasks are running slowly in a stage, they will be
re-launched.

你通过向spark-submit提供标志来打开它：

spark-submit  \
--conf "spark.speculation=true" \
--conf "spark.speculation.multiplier=5" \

有关您可以传递的不同设置的更多信息,请参阅Spark Configuration页面
>手动将HttpClient v4.5.1或更高版本传递给Sparks类路径,因此它可以在它的uber JAR中加载之前加载它.这可能有点困难,因为使用Spark的类加载过程有点麻烦.这意味着您可以按照以下方式执行以下操作：

CP=''; for f in /path/to/httpcomponents-client-4.5.2/lib/*.jar; do CP=$CP$f:; done
SPARK_CLASSPATH="$CP" sbin/start-master.sh   # on your master machine
SPARK_CLASSPATH="$CP" sbin/start-slave.sh 'spark://master_name:7077'

或者只是在spark-env.sh中将特定版本的JAR更新为SPARK_CLASSPATH.
>更新到Spark 2.0.0.新版本的Spark使用HttpClient v4.5.2解决了这个问题.

点赞