Cassandra Batch InvalidQueryException - Batch too large

Question

我正在使用Batch将数据插入Cassandra。当我运行作业时，我得到以下异常。

caused by: com.datastax.driver.core.exceptions.InvalidQueryException: Batch too large   
  at com.datastax.driver.core.Responses$Error.asException(Responses.java:136)   
  at com.datastax.driver.core.DefaultResultSetFuture.onSet(DefaultResultSetFuture.java:179)     
  at com.datastax.driver.core.RequestHandler.setFinalResult(RequestHandler.java:184)    
  at com.datastax.driver.core.RequestHandler.access$2500(RequestHandler.java:43)    
  at com.datastax.driver.core.RequestHandler$SpeculativeExecution.setFinalResult(RequestHandler.java:798)   
  at com.datastax.driver.core.RequestHandler$SpeculativeExecution.onSet(RequestHandler.java:617)    
  at com.datastax.driver.core.Connection$Dispatcher.channelRead0(Connection.java:1005)  
  at com.datastax.driver.core.Connection$Dispatcher.channelRead0(Connection.java:928)   
  at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)     
  at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)   
  at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)     
  at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:266)     
  at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)   
  at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)

我已经阅读了很多关于这个问题的博客。但这无济于事。我尝试在初始化时将"spark.cassandra.output.batch.size.bytes"设置为spark conf。这还没有解决我的问题。我收到同样的错误。我的批处理有大约 1000 个插入语句。

请在下面找到我的代码。

CassandraConnector connector = CassandraConnector.apply(javaSparkContext.getConf());  
pairRDD.mapToPair(earnCalculatorKeyIterableTuple2 -> {  
            if (condition) {  
                do something......
            }  
            else {  
                Session session = connector.openSession();  
                BatchStatement batch = new   BatchStatement(BatchStatement.Type.UNLOGGED);            batch.setConsistencyLevel(ConsistencyLevel.valueOf(LOCAL_QUOROM));  
                PreparedStatement statement = session.prepare('my insert query');  
                for (condition) {  
                    if (!condition) {  
                        break;  
                    }  
                    Tuple2._2.forEach(s -> {  
                        if (!condition) {  
                            LOG.info(message);  
                        }  
                        else {  
                            BoundStatement boundStatement = statement.bind("bind variables");  
                            batch.add(boundStatement);  
                        }  
                    });  
                    session.execute(batch);  
                    batch.clear();  
                }  
                session.close();  
            }  
            return Tuple2;  
        });  
        return s;  
    }  

感谢任何帮助。

Answer 1

您正在手动创建批处理，并且批处理太大。向每个批次添加较少的行。有很多方法可以手动执行此操作，但最简单的方法是添加一个计数器，每次添加 X 语句时都会提交一个批处理？

您要更改的参数仅与 saveToCassandra 完成的自动批处理有关。