我在 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”
我对分区下推的理解有误吗?我还缺少其他配置吗?
非常感谢您对此的投入。
谢谢!
最佳答案
单一存储凭据必须在所有节点上都相同才能利用分区下推。
如果您在所有节点上都具有相同的凭据,请尝试安装最新版本的 spark 连接器。
因为经常会因为 spark connector 和 singlestore 的兼容性问题而出现。
https://stackoverflow.com/questions/54882716/
相关文章:
python - Matplotlib png 输出在 reportlab 的 pdf 中出现 "b
Android WebRTC 不适用于不同的网络 - 无视频
javascript - 如何防止在 Angular 4+ 中更改查询字符串时页面刷新/重新加载?
google-chrome - 谷歌登录 "error": "popup_closed_by_use
javascript - OpenRouteService Directions api 用法
ubuntu-16.04 - 设置了 CATALINA_PID 但指定的文件不存在