C#批量插入40000行到Cassandra DB

Question

我想用batch向Cassandra插入40000行。但它总是停在32769，并给我一个异常"System.ArgumentOutOfRangeException“。我应该怎么做才能在Cassandra中插入超过32769行。

下面是我的代码：

        //建立DCS 資料
        DateTime ToDay = DateTime.Today;
        string LotStr = ToDay.ToString("yyMMdd");
        DateTime NowTime = DateTime.Now;            
        List<DCS_Model> DCS_list = new List<DCS_Model>();
        Random rnd = new Random();
        for (int i = 1; i <= 40000; i++)
        {
            DCS_list.Add(new DCS_Model(LotStr, String.Format("Tag_{0}", i), rnd.Next(1000) + rnd.NextDouble(), NowTime, NowTime));
        }

        //上傳至Cassandra
        DateTime tt = DateTime.Now;
        Cluster cluster = Cluster.Builder().AddContactPoint("192.168.52.182").Build();
        ISession session = cluster.Connect("testkeyspace");
        //List<PreparedStatement> StatementLs = new List<PreparedStatement>();
        var InsertDCS = session.Prepare("INSERT INTO DCS_Test (LOT, NAME, VALUE, CREATETIME, SERVERTIME) VALUES (?, ?, ?, ?, ?)");
        var batch = new BatchStatement();
        foreach (DCS_Model dcs in DCS_list)
        {

            batch.Add(InsertDCS.Bind(dcs.LOT,dcs.NAME,dcs.VALUE,dcs.CREATETIME,dcs.SERVERTIME));
        }
        session.Execute(batch);
        //Row result = session.Execute("select * from TestTable").First();
        TimeSpan CassandraTime = DateTime.Now - tt;
        //Console.WriteLine(CassandraTime);

当批量添加32768次时，它将在batch.Add(InsertDCS.Bind(dcs.LOT,dcs.NAME,dcs.VALUE,dcs.CREATETIME,dcs.SERVERTIME))停止。

请帮帮我。谢谢！！

Answer 1

RDBMS世界中的批处理功能甚至不能远程镜像Cassandra的批处理功能。它们的名称可能相同，但它们的设计目的不同。事实上，Cassandra的可能应该重命名为"atomic“，以避免混淆。

试着发送40k个单独的请求，与listenable异步(这样你就知道它们什么时候都完成了)，而不是一次性地将它们批处理在一起。我认为与Java的ListenableFuture相对应的C#是SettableFuture。你应该去查一查。

发送40,000个单独的事务可能看起来有悖于直觉。但它肯定比敲打一个Cassandra节点作为协调器(以及该节点将生成的所有网络流量)来处理和确保40k upsert的原子性要好。

此外，请确保使用令牌感知负载平衡策略。这将使您的upsert指向它所需的确切节点(为您节省了使用协调器的网络跳跃)。

Cluster cluster = Cluster.Builder().AddContactPoint("192.168.52.182").Build()
    .WithLoadBalancingPolicy(new TokenAwarePolicy
        (new DCAwareRoundRobinPolicy("westDC")));

Answer 2

我发现当add计数超过Int16.MaxValue时，函数"BatchStatement“的源代码会抛出异常。所以我修改了源代码，然后解决了这个问题！