火花工人连续运行给出错误

Question

我已经设置了 cassandra 集群，每个数据中心有 2 个数据中心和 3 个节点，每个数据中心有 1 个种子节点和一个复制因子 2。我的 spark 还设置有 2 个 2 核和 8 gb ram 的工作机器和一个主机带 4 GB 内存。现在我每小时运行一次 spark 作业，这个小时需要处理的数据约为 20,00,000。我的 Spark 作业持续运行，在我的工作节点中显示此错误

15/08/14 03:31:54 INFO LocalNodeFirstLoadBalancingPolicy: Suspected host 10.0.1.205 (DC2)
15/08/14 03:32:27 INFO RequestHandler: Query SELECT "addedtime" FROM "sams"."events" WHERE token("appname") > ? AND token("appname") <= ? AND "addedtime" >= ? AND "addedtime" < ?   ALLOW FILTERING is not prepared on /10.0.1.205:9042, preparing before retrying executing. Seeing this message a few times is fine, but seeing it a lot may be source of performance problems


 ERROR Session: Error creating pool to /10.0.1.205:9042
com.datastax.driver.core.ConnectionException: [/10.0.1.205:9042] Unexpected error during transport initialization (com.datastax.driver.core.OperationTimedOutException: [/10.0.1.205:9042] Operation timed out)
        at com.datastax.driver.core.Connection.initializeTransport(Connection.java:186)
        at com.datastax.driver.core.Connection.<init>(Connection.java:116)
        at com.datastax.driver.core.PooledConnection.<init>(PooledConnection.java:32)
        at com.datastax.driver.core.Connection$Factory.open(Connection.java:586)
        at com.datastax.driver.core.SingleConnectionPool.<init>(SingleConnectionPool.java:76)
        at com.datastax.driver.core.HostConnectionPool.newInstance(HostConnectionPool.java:35)
        at com.datastax.driver.core.SessionManager$2.call(SessionManager.java:231)
        at com.datastax.driver.core.SessionManager$2.call(SessionManager.java:224)
        at java.util.concurrent.FutureTask.run(FutureTask.java:262)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:745)
Caused by: com.datastax.driver.core.OperationTimedOutException: [/10.0.1.205:9042] Operation timed out
        at com.datastax.driver.core.Connection$Future.onTimeout(Connection.java:917)
        at com.datastax.driver.core.Connection$ResponseHandler$1.run(Connection.java:981)
        at org.jboss.netty.util.HashedWheelTimer$HashedWheelTimeout.expire(HashedWheelTimer.java:546)
        at org.jboss.netty.util.HashedWheelTimer$Worker.notifyExpiredTimeouts(HashedWheelTimer.java:446)
        at org.jboss.netty.util.HashedWheelTimer$Worker.run(HashedWheelTimer.java:395)
        at org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108)
        ... 1 more

请告诉我可能是什么问题。 以下是我运行 Spark 作业的选项

--conf spark.cassandra.input.split.size_in_mb=67108864
 --executor-memory 6G
--driver-memory 6G

添加我的表模式

CREATE TABLE events (
 addedtime timestamp,
 appname text,
 hostname text,
 assetname text,
 brandname text,
 eventname text,
 eventorigin text,
 eventtime timestamp,
 PRIMARY KEY ((appname), addedtime, hostname)
);

我的 appname 是分区键，addedtime 和主机名是集群键的一部分。

Answer 1

spark.cassandra.query.retry.delay

spark.cassandra.read.timeout_ms

spark.cassandra.input.split.size_in_mb

spark.cassandra.input.fetch.size_in_rows

你必须在 SparkConf 中设置。