将Dataframe写入面临busypoolexception的cassandra

ycggw6v2  于 2021-06-10  发布在  Cassandra
关注(0)|答案(1)|浏览(451)

我正试图用这行代码将Dataframe写入cassandra,本来有一天可以写入表,但突然出现了错误

alertdf
.write.format("org.apache.spark.sql.cassandra")
                 .options(Map("keyspace" -> "dummy", "table" -> "dummytable"))
                  .mode(SaveMode.Append)
                  .save()

我得到下面这个错误,无法找出是什么错了

ERROR QueryExecutor: Failed to execute: com.datastax.spark.connector.writer.RichBoundStatement@7dba59e2
        com.datastax.driver.core.exceptions.NoHostAvailableException: All host(s) tried for query failed (tried:**.**.**.**/**.**.**.**:9042 (com.datastax.driver.core.exceptions.BusyPoolException: [**.**.**.**/**.**.**.**] Pool is busy (no available connection and the queue has reached its max size 256)))
            at com.datastax.driver.core.RequestHandler.reportNoMoreHosts(RequestHandler.java:211)
            at com.datastax.driver.core.RequestHandler.access$1000(RequestHandler.java:46)
            at com.datastax.driver.core.RequestHandler$SpeculativeExecution.findNextHostAndQuery(RequestHandler.java:275)
            at com.datastax.driver.core.RequestHandler$SpeculativeExecution$1.onFailure(RequestHandler.java:338)
            at shade.com.datastax.spark.connector.google.common.util.concurrent.Futures$6.run(Futures.java:1310)
            at shade.com.datastax.spark.connector.google.common.util.concurrent.MoreExecutors$DirectExecutor.execute(MoreExecutors.java:457)
            at shade.com.datastax.spark.connector.google.common.util.concurrent.Futures$ImmediateFuture.addListener(Futures.java:106)
            at shade.com.datastax.spark.connector.google.common.util.concurrent.Futures.addCallback(Futures.java:1322)
            at shade.com.datastax.spark.connector.google.common.util.concurrent.Futures.addCallback(Futures.java:1258)
            at com.datastax.driver.core.RequestHandler$SpeculativeExecution.query(RequestHandler.java:297)
            at com.datastax.driver.core.RequestHandler$SpeculativeExecution.findNextHostAndQuery(RequestHandler.java:272)
            at com.datastax.driver.core.RequestHandler.startNewExecution(RequestHandler.java:115)
            at com.datastax.driver.core.RequestHandler.sendRequest(RequestHandler.java:95)
            at com.datastax.driver.core.SessionManager.executeAsync(SessionManager.java:132)
            at sun.reflect.GeneratedMethodAccessor34.invoke(Unknown Source)
            at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
            at java.lang.reflect.Method.invoke(Method.java:498)
            at com.datastax.spark.connector.cql.SessionProxy.invoke(SessionProxy.scala:40)
            at com.sun.proxy.$Proxy14.executeAsync(Unknown Source)
            at sun.reflect.GeneratedMethodAccessor34.invoke(Unknown Source)
            at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
            at java.lang.reflect.Method.invoke(Method.java:498)
            at com.datastax.spark.connector.cql.SessionProxy.invoke(SessionProxy.scala:40)
            at com.sun.proxy.$Proxy15.executeAsync(Unknown Source)
            at com.datastax.spark.connector.writer.QueryExecutor$$anonfun$$lessinit$greater$1.apply(QueryExecutor.scala:11)
            at com.datastax.spark.connector.writer.QueryExecutor$$anonfun$$lessinit$greater$1.apply(QueryExecutor.scala:11)
            at com.datastax.spark.connector.writer.AsyncExecutor.executeAsync(AsyncExecutor.scala:31)
            at com.datastax.spark.connector.writer.TableWriter$$anonfun$writeInternal$1$$anonfun$apply$2.apply(TableWriter.scala:199)
            at com.datastax.spark.connector.writer.TableWriter$$anonfun$writeInternal$1$$anonfun$apply$2.apply(TableWriter.scala:198)
            at scala.collection.Iterator$class.foreach(Iterator.scala:893)
            at com.datastax.spark.connector.writer.GroupingBatchBuilder.foreach(GroupingBatchBuilder.scala:31)
            at com.datastax.spark.connector.writer.TableWriter$$anonfun$writeInternal$1.apply(TableWriter.scala:198)
            at com.datastax.spark.connector.writer.TableWriter$$anonfun$writeInternal$1.apply(TableWriter.scala:175)
            at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$withSessionDo$1.apply(CassandraConnector.scala:112)
            at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$withSessionDo$1.apply(CassandraConnector.scala:111)
            at com.datastax.spark.connector.cql.CassandraConnector.closeResourceAfterUse(CassandraConnector.scala:145)
            at com.datastax.spark.connector.cql.CassandraConnector.withSessionDo(CassandraConnector.scala:111)
            at com.datastax.spark.connector.writer.TableWriter.writeInternal(TableWriter.scala:175)
            at com.datastax.spark.connector.writer.TableWriter.insert(TableWriter.scala:162)
            at com.datastax.spark.connector.writer.TableWriter.write(TableWriter.scala:149)
            at com.datastax.spark.connector.RDDFunctions$$anonfun$saveToCassandra$1.apply(RDDFunctions.scala:36)
            at com.datastax.spark.connector.RDDFunctions$$anonfun$saveToCassandra$1.apply(RDDFunctions.scala:36)
            at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
            at org.apache.spark.scheduler.Task.run(Task.scala:86)
            at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
            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)

有人能帮我解决这个问题吗?

irtuqstp

irtuqstp1#

看起来服务器过载了,并且不按时处理请求。我建议尝试调整与写相关的配置参数,比如, output.concurrent.writes , output.throughput_mb_per_sec 其他的,但我会从前两个开始。

相关问题