You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ja...@apache.org on 2017/08/10 11:46:00 UTC
carbondata git commit: [CARBONDATA-1366]add an option
'carbon.global.sort.rdd.storage.level' to configure rdd storage level when
sort_scope=global_sort
Repository: carbondata
Updated Branches:
refs/heads/master 7b5a1c3b9 -> 0be69cd36
[CARBONDATA-1366]add an option 'carbon.global.sort.rdd.storage.level' to configure rdd storage level when sort_scope=global_sort
add an option 'carbon.global.sort.rdd.storage.level' for users to configure rdd storage level according to their different env, and the default value of this option is MEMORY_ONLY.
This closes #1245
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/0be69cd3
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/0be69cd3
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/0be69cd3
Branch: refs/heads/master
Commit: 0be69cd36cd875b35499ea78efb0297bf0377a93
Parents: 7b5a1c3
Author: Zhang Zhichao <44...@qq.com>
Authored: Tue Aug 8 17:42:40 2017 +0800
Committer: Jacky Li <ja...@qq.com>
Committed: Thu Aug 10 19:45:37 2017 +0800
----------------------------------------------------------------------
.../core/constants/CarbonCommonConstants.java | 16 +++++++++++
.../carbondata/core/util/CarbonProperties.java | 20 +++++++++++++-
.../apache/carbondata/core/util/CarbonUtil.java | 28 ++++++++++++++++++++
.../load/DataLoadProcessBuilderOnSpark.scala | 4 ++-
4 files changed, 66 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0be69cd3/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
index c9f9373..f06eb83 100644
--- a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
+++ b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
@@ -1304,6 +1304,22 @@ public final class CarbonCommonConstants {
*/
public static final String CARBON_USE_MULTI_TEMP_DIR_DEFAULT = "false";
+ /**
+ * Which storage level to persist rdd when sort_scope=global_sort
+ */
+ @CarbonProperty
+ public static final String CARBON_GLOBAL_SORT_RDD_STORAGE_LEVEL =
+ "carbon.global.sort.rdd.storage.level";
+
+ /**
+ * The default value(MEMORY_ONLY) is designed for executors with big memory, if user's executor
+ * has less memory, set the CARBON_GLOBAL_SORT_RDD_STORAGE_LEVEL to MEMORY_AND_DISK_SER or
+ * other storage level to correspond to different environment.
+ * You can get more recommendations about storage level in spark website:
+ * http://spark.apache.org/docs/latest/rdd-programming-guide.html#rdd-persistence.
+ */
+ public static final String CARBON_GLOBAL_SORT_RDD_STORAGE_LEVEL_DEFAULT = "MEMORY_ONLY";
+
private CarbonCommonConstants() {
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0be69cd3/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java
index 12776de..2620ecb 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java
@@ -867,7 +867,7 @@ public final class CarbonProperties {
CarbonCommonConstants.CARBON_USE_MULTI_TEMP_DIR_DEFAULT);
boolean validateBoolean = CarbonUtil.validateBoolean(usingMultiDirStr);
if (!validateBoolean) {
- LOGGER.info("The carbon.use.multiple.temp.dir configuration value is invalid."
+ LOGGER.error("The carbon.use.multiple.temp.dir configuration value is invalid."
+ "Configured value: \"" + usingMultiDirStr + "\"."
+ "Data Load will not use multiple temp directories.");
usingMultiDirStr = CarbonCommonConstants.CARBON_USE_MULTI_TEMP_DIR_DEFAULT;
@@ -876,6 +876,24 @@ public final class CarbonProperties {
}
/**
+ * Return valid storage level
+ * @return String
+ */
+ public String getGlobalSortRddStorageLevel() {
+ String storageLevel = getProperty(CarbonCommonConstants.CARBON_GLOBAL_SORT_RDD_STORAGE_LEVEL,
+ CarbonCommonConstants.CARBON_GLOBAL_SORT_RDD_STORAGE_LEVEL_DEFAULT);
+ boolean validateStorageLevel = CarbonUtil.isValidStorageLevel(storageLevel);
+ if (!validateStorageLevel) {
+ LOGGER.error("The " + CarbonCommonConstants.CARBON_GLOBAL_SORT_RDD_STORAGE_LEVEL
+ + " configuration value is invalid. It will use default storage level("
+ + CarbonCommonConstants.CARBON_GLOBAL_SORT_RDD_STORAGE_LEVEL_DEFAULT
+ + ") to persist rdd.");
+ storageLevel = CarbonCommonConstants.CARBON_GLOBAL_SORT_RDD_STORAGE_LEVEL_DEFAULT;
+ }
+ return storageLevel.toUpperCase();
+ }
+
+ /**
* returns true if carbon property
* @param key
* @return
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0be69cd3/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
index 7628415..edc4c28 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
@@ -1714,6 +1714,34 @@ public final class CarbonUtil {
}
/**
+ * validate the storage level
+ * @param storageLevel
+ * @return boolean
+ */
+ public static boolean isValidStorageLevel(String storageLevel) {
+ if (null == storageLevel || storageLevel.trim().equals("")) {
+ return false;
+ }
+ switch (storageLevel.toUpperCase()) {
+ case "DISK_ONLY":
+ case "DISK_ONLY_2":
+ case "MEMORY_ONLY":
+ case "MEMORY_ONLY_2":
+ case "MEMORY_ONLY_SER":
+ case "MEMORY_ONLY_SER_2":
+ case "MEMORY_AND_DISK":
+ case "MEMORY_AND_DISK_2":
+ case "MEMORY_AND_DISK_SER":
+ case "MEMORY_AND_DISK_SER_2":
+ case "OFF_HEAP":
+ case "NONE":
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ /**
* validate teh batch size
*
* @param value
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0be69cd3/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala
index 534ab88..fed8a96 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala
@@ -32,6 +32,7 @@ import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.row.CarbonRow
import org.apache.carbondata.core.statusmanager.LoadMetadataDetails
+import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.processing.csvload.{CSVInputFormat, StringArrayWritable}
import org.apache.carbondata.processing.model.CarbonLoadModel
import org.apache.carbondata.processing.newflow.DataLoadProcessBuilder
@@ -112,7 +113,8 @@ object DataLoadProcessBuilderOnSpark {
// Because if the number of partitions greater than 1, there will be action operator(sample) in
// sortBy operator. So here we cache the rdd to avoid do input and convert again.
if (numPartitions > 1) {
- convertRDD.persist(StorageLevel.MEMORY_AND_DISK)
+ convertRDD.persist(StorageLevel.fromString(
+ CarbonProperties.getInstance().getGlobalSortRddStorageLevel()))
}
import scala.reflect.classTag