如何使用Spark在Cassandra中列出大型Cassandra表中的所有分区键？

Question

我们有一个叫做cassandra扫描的程序，它使用spark-cassandra连接器在一个非常大的表中列出分区键的所有值。该表有大约1700万个Cassandra分区，每个分区平均有200行。包含该表的Cassandra集群在6个节点上运行DSE 5.1.8。包含该表的键空间的复制因子为3。

以下是密钥空间和表的简化定义。

CREATE KEYSPACE myspace WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3'}  AND durable_writes = true;

CREATE TABLE myspace.largetable (
    id text,
    itemOrder text,
    ...
    PRIMARY KEY (id, itemOrder)
) WITH CLUSTERING ORDER BY (itemOrder ASC)

cassandra-scan中用于列出分区键的所有值的语句如下:

val res = sc.cassandraTable(keyspace, table).select("id").perPartitionLimit(1).repartition(320)

我们使用Apache Spark 2.3.1和spark-cassandra-connector 2.3.2。用于启动cassandra-scan的命令如下。

/path/to/spark/spark-2.3.1-bin-hadoop2.7/bin/spark-submit --class "CassandraScan" --jars /path/to/spark-cassandra-connector_2.11-2.3.2.jar --executor-memory 15g --master local[20] cassandra-scan.jar &

cassandra-scan运行正常，大约需要19个小时。

我们最近建立了一个新的 Cassandra 集群，同样有 6 个节点（与第一个集群中使用的节点不同）。此群集运行 DSE 6.8.16。第一个表中的所有数据都已添加到新集群中的表中。

我们将Apache Spark的版本更新为2.4.8，将Spark cassandra连接器更新为2.4.2。我们测试了Spark分区数在2000到200000之间的程序。我们无法让cassandr扫描正常运行。我们看到以下格式的错误：

java.io.IOException: Exception during execution of SELECT "id" FROM "myspace"."largetable" WHERE token("id") > ? AND token("id") <= ?  PER PARTITION LIMIT 1 ALLOW FILTERING: Not enough replicas available for query at consistency LOCAL_ONE (1 required but only 0 alive)

一些 cassandra-scan 运行导致一些 Cassandra 节点关闭，并在 Cassandra 日志中显示如下消息。

INFO  [CoreThread-22] 2022-04-03 06:26:35,467  InboundHandshakeHandler.java:353 - Failed to properly handshake with peer /xxx.xxx.xxx.xxx:41231. Closing the channel.
java.lang.OutOfMemoryError: Direct buffer memory

WARN  [Outbound-/xxx.xxx.xxx.xxx-large-message-writer] 2022-04-01 19:17:58,248  AbstractOutboundMessageHandler.java:80 - LARGE_MESSAGE with id 97 from /xxx.xxx.xxx.xxx to /xxx.xxx.xxx.xxx via (/xxx.xxx.xxx.xxx,/xxx.xxx.xxx.xxx:7000) error...
java.io.IOException: java.lang.RuntimeException: io.netty.channel.unix.Errors$NativeIoException: writeAddress(..) failed: Connection reset by peer

非常感谢任何帮助让这个工作。谢谢。

Answer 1

我们使用DataStax Bulk Loader解决了这个问题。

dsbulk unload \
  --connector.csv.url <path>/<to>/<outputDir> \
  -h <host> \
  -query "select distinct id from myspace.largetable"

dsbulk花费了大约3个小时来获得1750万个值。

Answer 2

此错误表示群集中至少有一个节点无法为请求提供服务：

    Not enough replicas available for query at consistency LOCAL_ONE \
      (1 required but only 0 alive)

您需要查看Cassandra日志来确定(1)哪个节点没有响应/不可用，以及(2)原因。干杯！