java - 使用 spark 对抗 cassandra 性能不佳

Question

目前，我们在对我们的 cassandra 数据库使用 apache spark 时挑战我们的架构，因为我们遇到了非常糟糕的读取性能。

spark & cassandra 发生的硬件是一个云服务器，16GB 内存，8 核，操作系统使用 SSD。

Cassandra 'data_file_directories' 设置为另一个硬盘，其测试结果为 hdparm -tT:

Timing cached reads:   13140 MB in  1.99 seconds = 6604.42 MB/sec
Timing buffered disk reads: 428 MB in  3.00 seconds = 142.65 MB/sec

cassandra中的 objective-c f:

CREATE TABLE test.stats (
day timestamp,
received timestamp,
target inet,
via inet,
prefix blob,
rtt decimal,
PRIMARY KEY (day, received, target, via)
) WITH CLUSTERING ORDER BY (received ASC, target ASC, via ASC)
    AND bloom_filter_fp_chance = 0.01
    AND caching = '{"keys":"NONE", "rows_per_partition":"NONE"}'
    AND comment = ''
    AND compaction = {'min_threshold': '4', 'class':     'org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy',     'max_threshold': '32'}
    AND compression = {}
    AND dclocal_read_repair_chance = 0.1
    AND default_time_to_live = 0
    AND gc_grace_seconds = 864000
    AND max_index_interval = 2048
    AND memtable_flush_period_in_ms = 0
    AND min_index_interval = 128
    AND read_repair_chance = 0.0
    AND speculative_retry = '99.0PERCENTILE';

我们目前使用的是 Cassandra 2.1.6 以及 java datastax 驱动程序 (2.1.6) 和 spark java 连接器(spark-cassandra-connector_2.10，版本 1.4.0-M2)。

Spark 进程目前有一个 worker 设置了 conf.set("spark.executor.memory", "2G");。

启动一个简单的 spark 作业以通过提交/或序列化驱动程序读取一个显式分区键(大约有 83.520.000 行)的所有行需要 17 分钟。

该作业只是将所有行写入一个文件，该文件的最终大小为 1,2G。

驱动代码如下:

CassandraTableScanJavaRDD<EchoRepliesBean> cassandraTable2 = null;
    switch (timespanMode)
    {
        case SIX_HOURS:
            Calendar calendarDay = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
            calendarDay.setTimeInMillis(now);
            calendarDay.set(Calendar.HOUR_OF_DAY, 0);
            calendarDay.set(Calendar.MINUTE, 0);
            calendarDay.set(Calendar.SECOND, 0);
            calendarDay.set(Calendar.MILLISECOND, 0);
            Timestamp tsEnd = new Timestamp(calendarDay.getTimeInMillis());
            calendarDay.add(Calendar.DAY_OF_MONTH, -1);
            Timestamp tsStart = new Timestamp(calendarDay.getTimeInMillis());
            System.out.println(tsStart);
            cassandraTable2 = javaFunctions(sc).cassandraTable("test", "stats", mapRowTo(EchoRepliesBean.class))
                                               .where("day = ?", tsStart);
            break;
        default:
            /* make compiler happy */
            // cassandraTable = null;
    } 
    cassandraTable2.saveAsTextFile("/opt/out_TEST_" + System.currentTimeMillis());

    sc.stop();

这很奇怪，对于进一步调试的任何帮助或想法将不胜感激。

Answer 1

提高性能的一些可能方法:

1. 通过在多个节点上对数据进行分区来提高并行度。由于您按天进行分区，因此一个节点上的一个分区中有大量行。这迫使读取和写入成为串行操作。如果您按小时进行分区，那么您的数据可能会分布在多个节点和多个 Spark Worker 中。

2. 我怀疑您的 day 分区太大，无法放入单个 spark worker 的内存，这可能会导致一些数据交换到磁盘。使用较小的分区、为 spark worker 提供更多内存或使用更多 spark worker 可以避免这种情况。

3. 确保您的 spark worker 在 Cassandra 节点上运行，而不是在单独的机器上运行。如果工作人员在不同的机器上，那么将数据从节点转移到工作人员将产生大量网络开销。

4. 确保您的云服务器为 Cassandra 使用本地存储而不是网络存储。

为了调试，我会尝试在其中只有一行的分区上运行您的测试。如果表现不佳，则说明您的机器设置有问题。如果性能良好，则增加分区中的行数，直到您看到性能急剧下降。