You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ra...@apache.org on 2018/08/09 18:26:25 UTC
[41/47] carbondata git commit: [CARBONDATA-2659] Support partition
table by DataFrame API
[CARBONDATA-2659] Support partition table by DataFrame API
Currently only partition table is only supported by SQL, it should be supported by Spark DataFrame API also.
This PR added an option to specify the partition columns when writing a DataFrame to carbon table.
This closes #2415
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/7598dea8
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/7598dea8
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/7598dea8
Branch: refs/heads/branch-1.4
Commit: 7598dea8ef2327557e580a59566932642d64fbf5
Parents: 7535d46
Author: Jacky Li <ja...@qq.com>
Authored: Tue Jun 26 20:27:37 2018 +0800
Committer: ravipesala <ra...@gmail.com>
Committed: Thu Aug 9 23:51:36 2018 +0530
----------------------------------------------------------------------
.../examples/CarbonDataFrameExample.scala | 7 +--
.../StandardPartitionTableQueryTestCase.scala | 21 +++++++++
.../apache/carbondata/spark/CarbonOption.scala | 45 ++++++++++++--------
.../spark/sql/CarbonDataFrameWriter.scala | 35 +++++++++++++--
4 files changed, 84 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7598dea8/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonDataFrameExample.scala
----------------------------------------------------------------------
diff --git a/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonDataFrameExample.scala b/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonDataFrameExample.scala
index 65c3b78..2abd151 100644
--- a/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonDataFrameExample.scala
+++ b/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonDataFrameExample.scala
@@ -33,20 +33,21 @@ object CarbonDataFrameExample {
// Writes Dataframe to CarbonData file:
import spark.implicits._
val df = spark.sparkContext.parallelize(1 to 100)
- .map(x => ("a", "b", x))
+ .map(x => ("a" + x % 10, "b", x))
.toDF("c1", "c2", "number")
// Saves dataframe to carbondata file
df.write
.format("carbondata")
.option("tableName", "carbon_df_table")
- .option("compress", "true")
- .option("tempCSV", "false")
+ .option("partitionColumns", "c1") // a list of column names
.mode(SaveMode.Overwrite)
.save()
spark.sql(""" SELECT * FROM carbon_df_table """).show()
+ spark.sql("SHOW PARTITIONS carbon_df_table").show()
+
// Specify schema
import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}
val customSchema = StructType(Array(
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7598dea8/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableQueryTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableQueryTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableQueryTestCase.scala
index fb554f2..eaf5e4b 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableQueryTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableQueryTestCase.scala
@@ -73,6 +73,27 @@ class StandardPartitionTableQueryTestCase extends QueryTest with BeforeAndAfterA
}
+ test("create partition table by dataframe") {
+ sql("select * from originTable")
+ .write
+ .format("carbondata")
+ .option("tableName", "partitionxxx")
+ .option("partitionColumns", "empno")
+ .save("Overwrite")
+
+ val frame = sql(
+ "select empno, empname, designation, doj, workgroupcategory, workgroupcategoryname, deptno," +
+ " deptname, projectcode, projectjoindate, projectenddate, attendance, utilization, salary " +
+ "from partitionxxx where empno=11 order by empno")
+ verifyPartitionInfo(frame, Seq("empno=11"))
+
+ checkAnswer(frame,
+ sql("select empno, empname, designation, doj, workgroupcategory, workgroupcategoryname, deptno, " +
+ "deptname, projectcode, projectjoindate, projectenddate, attendance, utilization, salary " +
+ "from originTable where empno=11 order by empno"))
+ sql("drop table if exists partitionxxx")
+ }
+
test("querying on partition table for string partition column") {
sql(
"""
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7598dea8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/CarbonOption.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/CarbonOption.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/CarbonOption.scala
index 9724fa8..3acfb9c 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/CarbonOption.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/CarbonOption.scala
@@ -23,38 +23,49 @@ package org.apache.carbondata.spark
*/
class CarbonOption(options: Map[String, String]) {
- def dbName: Option[String] = options.get("dbName")
+ lazy val dbName: Option[String] = options.get("dbName")
- def tableName: String = options.getOrElse("tableName", "default_table")
+ lazy val tableName: String = options.getOrElse("tableName", "default_table")
- def tablePath: Option[String] = options.get("tablePath")
+ lazy val tablePath: Option[String] = options.get("tablePath")
- def partitionCount: String = options.getOrElse("partitionCount", "1")
+ lazy val partitionCount: String = options.getOrElse("partitionCount", "1")
- def tempCSV: Boolean = options.getOrElse("tempCSV", "false").toBoolean
+ lazy val partitionClass: String = {
+ options.getOrElse("partitionClass",
+ "org.apache.carbondata.processing.partition.impl.SampleDataPartitionerImpl")
+ }
+
+ lazy val compress: Boolean = options.getOrElse("compress", "false").toBoolean
- def compress: Boolean = options.getOrElse("compress", "false").toBoolean
+ lazy val partitionColumns: Option[Seq[String]] = {
+ if (options.contains("partitionColumns")) {
+ Option(options("partitionColumns").split(",").map(_.trim))
+ } else {
+ None
+ }
+ }
- def singlePass: Boolean = options.getOrElse("single_pass", "false").toBoolean
+ lazy val singlePass: Boolean = options.getOrElse("single_pass", "false").toBoolean
- def sortColumns: Option[String] = options.get("sort_columns")
+ lazy val sortColumns: Option[String] = options.get("sort_columns")
- def dictionaryInclude: Option[String] = options.get("dictionary_include")
+ lazy val dictionaryInclude: Option[String] = options.get("dictionary_include")
- def dictionaryExclude: Option[String] = options.get("dictionary_exclude")
+ lazy val dictionaryExclude: Option[String] = options.get("dictionary_exclude")
- def longStringColumns: Option[String] = options.get("long_string_columns")
+ lazy val longStringColumns: Option[String] = options.get("long_string_columns")
- def tableBlockSize: Option[String] = options.get("table_blocksize")
+ lazy val tableBlockSize: Option[String] = options.get("table_blocksize")
- def bucketNumber: Int = options.getOrElse("bucketnumber", "0").toInt
+ lazy val bucketNumber: Int = options.getOrElse("bucketnumber", "0").toInt
- def bucketColumns: String = options.getOrElse("bucketcolumns", "")
+ lazy val bucketColumns: String = options.getOrElse("bucketcolumns", "")
- def isBucketingEnabled: Boolean = options.contains("bucketcolumns") &&
+ lazy val isBucketingEnabled: Boolean = options.contains("bucketcolumns") &&
options.contains("bucketnumber")
- def isStreaming: Boolean = {
+ lazy val isStreaming: Boolean = {
var stream = options.getOrElse("streaming", "false")
if (stream.equalsIgnoreCase("sink") || stream.equalsIgnoreCase("source")) {
stream = "true"
@@ -62,7 +73,7 @@ class CarbonOption(options: Map[String, String]) {
stream.toBoolean
}
- def overwriteEnabled: Boolean =
+ lazy val overwriteEnabled: Boolean =
options.getOrElse("overwrite", "false").toBoolean
def toMap: Map[String, String] = options
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7598dea8/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala
index c81622e..6ead03a 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala
@@ -21,6 +21,7 @@ import org.apache.spark.sql.execution.command.management.CarbonLoadDataCommand
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.CarbonException
+import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.metadata.datatype.{DataTypes => CarbonType}
import org.apache.carbondata.spark.CarbonOption
@@ -78,10 +79,6 @@ class CarbonDataFrameWriter(sqlContext: SQLContext, val dataFrame: DataFrame) {
}
private def makeCreateTableString(schema: StructType, options: CarbonOption): String = {
- val carbonSchema = schema.map { field =>
- s"${ field.name } ${ convertToCarbonType(field.dataType) }"
- }
-
val property = Map(
"SORT_COLUMNS" -> options.sortColumns,
"DICTIONARY_INCLUDE" -> options.dictionaryInclude,
@@ -92,11 +89,41 @@ class CarbonDataFrameWriter(sqlContext: SQLContext, val dataFrame: DataFrame) {
).filter(_._2.isDefined)
.map(property => s"'${property._1}' = '${property._2.get}'").mkString(",")
+ val partition: Seq[String] = if (options.partitionColumns.isDefined) {
+ if (options.partitionColumns.get.toSet.size != options.partitionColumns.get.length) {
+ throw new MalformedCarbonCommandException(s"repeated partition column")
+ }
+ options.partitionColumns.get.map { column =>
+ val field = schema.fields.find(_.name.equalsIgnoreCase(column))
+ if (field.isEmpty) {
+ throw new MalformedCarbonCommandException(s"invalid partition column: $column")
+ }
+ s"$column ${field.get.dataType.typeName}"
+ }
+ } else {
+ Seq()
+ }
+
+ val schemaWithoutPartition = if (options.partitionColumns.isDefined) {
+ val partitionCols = options.partitionColumns.get
+ val fields = schema.filterNot {
+ field => partitionCols.exists(_.equalsIgnoreCase(field.name))
+ }
+ StructType(fields)
+ } else {
+ schema
+ }
+
+ val carbonSchema = schemaWithoutPartition.map { field =>
+ s"${ field.name } ${ convertToCarbonType(field.dataType) }"
+ }
+
val dbName = CarbonEnv.getDatabaseName(options.dbName)(sqlContext.sparkSession)
s"""
| CREATE TABLE IF NOT EXISTS $dbName.${options.tableName}
| (${ carbonSchema.mkString(", ") })
+ | ${ if (partition.nonEmpty) s"PARTITIONED BY (${partition.mkString(", ")})" else ""}
| STORED BY 'carbondata'
| ${ if (options.tablePath.nonEmpty) s"LOCATION '${options.tablePath.get}'" else ""}
| ${ if (property.nonEmpty) "TBLPROPERTIES (" + property + ")" else "" }