卡桑德拉的execute_concurrent不能正常工作

Question

我编写了一个Python脚本，它读取csv并使用异步和并发但并发的方式将它们插入Cassandra表中，但并发比异步慢。我使用并发的目的是实现并行写入，因此，加快在cassandra中索引csv文件的任务。

使用异步的代码：

for df in chunks:
            futures = []
            df = df.to_dict(orient='records')
            chunk_counter += 1
            for row in df:
                key = str(row["0"])
                row = json.dumps(row, default=str)
                futures.append(
                    self.session.execute_async(
                        insert_sql, [key, "version_1", row]
                    )
                )
                # batch.add(insert_sql, (key, "version_1", row))
            # self.session.execute(batch)
            for future in futures:
                self.log.debug(future)
                continue

使用并发代码：

for df in chunks:
            futures = []
            df = df.to_dict(orient='records')
            chunk_counter += 1
            for row in df:
                key = str(row["0"])
                row = json.dumps(row, default=str)
                params = (key, row, )
                futures.append(
                    (
                        insert_sql,
                        params
                    )
                )
            results = execute_concurrent(
                self.session, futures, raise_on_first_error=False)
            for (success, result) in results:
                if not success:
                    self.handle_error(result)  # result will be an Exception

Answer 1

您没有设置execute_concurrent的execute_concurrent参数，默认情况下它使用100。

来自文档

并发参数控制并发执行语句的数量。当Cluster.protocol_version设置为1或2时，建议将其保持在每台主机的核心连接数乘以已连接主机数量的100倍以下(参见Cluster.set_core_connections_per_host())。如果超出该数量，则事件循环线程可能试图阻止新连接的创建，从而严重影响吞吐量。如果protocol_version是3或更高版本，则可以安全地尝试使用更高级别的并发性。