scala - 使用 MemSQL 中的分区下推在 Spark 中启用并行化

Question

我在 MemSQL 中有一个 columnstore 表，它具有类似于下面的架构:

CREATE TABLE key_metrics (
source_id TEXT,
date TEXT,
metric1 FLOAT,
metric2 FLOAT,
…
SHARD KEY (source_id, date) USING CLUSTERED COLUMNSTORE
);

我有一个查询 MemSQL 表的 Spark 应用程序(与 Spark 作业服务器一起运行)。下面是我正在执行的那种 Dataframe 操作的简化形式(在 Scala 中):

sparkSession
.read
.format(“com.memsql.spark.connector”)
.options( Map (“path” -> “dbName.key_metrics”))
.load()
.filter(col(“source_id”).equalTo(“12345678”)
.filter(col(“date”)).isin(Seq(“2019-02-01”, “2019-02-02”, “2019-02-03”))

我已经通过查看物理计划确认这些过滤谓词正在被推送到 MemSQL。

我还检查了表中分区的分布是否相当均匀:

±--------------±----------------±-------------±-------±-----------+
| DATABASE_NAME | TABLE_NAME | PARTITION_ID | ROWS | MEMORY_USE |
±--------------±----------------±-------------±-------±-----------+
| dbName        | key_metrics |           0 | 784012 |        0 |
| dbName        | key_metrics |           1 | 778441 |        0 |
| dbName        | key_metrics |           2 | 671606 |        0 |
| dbName        | key_metrics |           3 | 748569 |        0 |
| dbName        | key_metrics |           4 | 622241 |        0 |
| dbName        | key_metrics |           5 | 739029 |        0 |
| dbName        | key_metrics |           6 | 955205 |        0 |
| dbName        | key_metrics |           7 | 751677 |        0 |
±--------------±----------------±-------------±-------±-----------+

我的问题是关于分区下推。据我了解，有了它，我们可以使用机器的所有核心并利用并行性进行批量加载。根据文档，这是通过创建与 MemSQL 数据库分区一样多的 Spark 任务来完成的。

但是，当运行 Spark 管道并观察 Spark UI 时，似乎只创建了一个 Spark 任务，它对在单个内核上运行的数据库进行单个查询。

我已确保还设置了以下属性:

spark.memsql.disablePartitionPushdown = false

spark.memsql.defaultDatabase = “dbName”

我对分区下推的理解有误吗？我还缺少其他配置吗？

非常感谢您对此的投入。

谢谢!

Answer 1

单一存储凭据必须在所有节点上都相同才能利用分区下推。

如果您在所有节点上都具有相同的凭据，请尝试安装最新版本的 spark 连接器。

因为经常会因为 spark connector 和 singlestore 的兼容性问题而出现。