This is an automated email from the ASF dual-hosted git repository.
zhaoc pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-doris.git
The following commit(s) were added to refs/heads/master by this push:
new 1550401 Support param exec_mem_limit for spark-doris-connctor (#2775)
1550401 is described below
commit 1550401d4b8b66ab3cc49f3a35269417b33a439a
Author: Youngwb <[email protected]>
AuthorDate: Sat Jan 18 00:14:39 2020 +0800
Support param exec_mem_limit for spark-doris-connctor (#2775)
---
be/src/runtime/fragment_mgr.cpp | 1 +
extension/spark-doris-connector/README.md | 1 +
.../java/org/apache/doris/spark/cfg/ConfigurationOptions.java | 3 +++
.../main/scala/org/apache/doris/spark/rdd/ScalaValueReader.scala | 9 +++++++++
.../src/main/thrift/doris/DorisExternalService.thrift | 3 +++
gensrc/thrift/DorisExternalService.thrift | 3 +++
6 files changed, 20 insertions(+)
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index 7deb819..a60dc47 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -665,6 +665,7 @@ Status FragmentMgr::exec_external_plan_fragment(const
TScanOpenParams& params, c
TQueryOptions query_options;
query_options.batch_size = params.batch_size;
query_options.query_timeout = params.query_timeout;
+ query_options.mem_limit = params.mem_limit;
query_options.query_type = TQueryType::EXTERNAL;
exec_fragment_params.__set_query_options(query_options);
VLOG_ROW << "external exec_plan_fragment params is "
diff --git a/extension/spark-doris-connector/README.md
b/extension/spark-doris-connector/README.md
index 2b0d2ef..d32db83 100644
--- a/extension/spark-doris-connector/README.md
+++ b/extension/spark-doris-connector/README.md
@@ -101,6 +101,7 @@ dorisSparkRDD.collect()
| doris.request.query.timeout.s | 3600 |
查询doris的超时时间，默认值为1小时，-1表示无超时限制 |
| doris.request.tablet.size | Integer.MAX_VALUE | 一个RDD
Partition对应的Doris Tablet个数。<br />此数值设置越小，则会生成越多的Partition。<br
/>从而提升Spark侧的并行度，但同时会对Doris造成更大的压力。 |
| doris.batch.size | 1024 | 一次从BE读取数据的最大行数。<br
/>增大此数值可减少Spark与Doris之间建立连接的次数。<br />从而减轻网络延迟所带来的的额外时间开销。 |
+| doris.exec.mem.limit | 2147483648 | 单个查询的内存限制。默认为
2GB，单位为字节 |
### SQL and Dataframe Only
diff --git
a/extension/spark-doris-connector/src/main/java/org/apache/doris/spark/cfg/ConfigurationOptions.java
b/extension/spark-doris-connector/src/main/java/org/apache/doris/spark/cfg/ConfigurationOptions.java
index d9b4231..742c3eb 100644
---
a/extension/spark-doris-connector/src/main/java/org/apache/doris/spark/cfg/ConfigurationOptions.java
+++
b/extension/spark-doris-connector/src/main/java/org/apache/doris/spark/cfg/ConfigurationOptions.java
@@ -53,5 +53,8 @@ public interface ConfigurationOptions {
String DORIS_BATCH_SIZE = "doris.batch.size";
int DORIS_BATCH_SIZE_DEFAULT = 1024;
+ String DORIS_EXEC_MEM_LIMIT = "doris.exec.mem.limit";
+ long DORIS_EXEC_MEM_LIMIT_DEFAULT = 2147483648L;
+
String DORIS_VALUE_READER_CLASS = "doris.value.reader.class";
}
diff --git
a/extension/spark-doris-connector/src/main/scala/org/apache/doris/spark/rdd/ScalaValueReader.scala
b/extension/spark-doris-connector/src/main/scala/org/apache/doris/spark/rdd/ScalaValueReader.scala
index 163ee16..13a955a 100644
---
a/extension/spark-doris-connector/src/main/scala/org/apache/doris/spark/rdd/ScalaValueReader.scala
+++
b/extension/spark-doris-connector/src/main/scala/org/apache/doris/spark/rdd/ScalaValueReader.scala
@@ -71,8 +71,16 @@ class ScalaValueReader(partition: PartitionDefinition,
settings: Settings) {
DORIS_REQUEST_QUERY_TIMEOUT_S_DEFAULT
}
+ val execMemLimit = Try {
+ settings.getProperty(DORIS_EXEC_MEM_LIMIT,
DORIS_EXEC_MEM_LIMIT_DEFAULT.toString).toLong
+ } getOrElse {
+ logger.warn(ErrorMessages.PARSE_NUMBER_FAILED_MESSAGE,
DORIS_EXEC_MEM_LIMIT, settings.getProperty(DORIS_EXEC_MEM_LIMIT))
+ DORIS_EXEC_MEM_LIMIT_DEFAULT
+ }
+
params.setBatch_size(batchSize)
params.setQuery_timeout(queryDorisTimeout)
+ params.setMem_limit(execMemLimit)
params.setUser(settings.getProperty(DORIS_REQUEST_AUTH_USER, ""))
params.setPasswd(settings.getProperty(DORIS_REQUEST_AUTH_PASSWORD, ""))
@@ -83,6 +91,7 @@ class ScalaValueReader(partition: PartitionDefinition,
settings: Settings) {
s"tabletId: ${params.getTablet_ids}, " +
s"batch size: $batchSize, " +
s"query timeout: $queryDorisTimeout, " +
+ s"execution memory limit: $execMemLimit, " +
s"user: ${params.getUser}, " +
s"query plan: ${params.opaqued_query_plan}")
diff --git
a/extension/spark-doris-connector/src/main/thrift/doris/DorisExternalService.thrift
b/extension/spark-doris-connector/src/main/thrift/doris/DorisExternalService.thrift
index d3f7e8e..c169874 100644
---
a/extension/spark-doris-connector/src/main/thrift/doris/DorisExternalService.thrift
+++
b/extension/spark-doris-connector/src/main/thrift/doris/DorisExternalService.thrift
@@ -56,6 +56,9 @@ struct TScanOpenParams {
11: optional i16 keep_alive_min
12: optional i32 query_timeout
+
+ // memory limit for a single query
+ 13: optional i64 mem_limit
}
struct TScanColumnDesc {
diff --git a/gensrc/thrift/DorisExternalService.thrift
b/gensrc/thrift/DorisExternalService.thrift
index d3f7e8e..483df19 100644
--- a/gensrc/thrift/DorisExternalService.thrift
+++ b/gensrc/thrift/DorisExternalService.thrift
@@ -56,6 +56,9 @@ struct TScanOpenParams {
11: optional i16 keep_alive_min
12: optional i32 query_timeout
+
+ // memory limit for a single query
+ 13: optional i64 mem_limit
}
struct TScanColumnDesc {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
