This is an automated email from the ASF dual-hosted git repository.
kunalkapoor pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/carbondata.git
The following commit(s) were added to refs/heads/master by this push:
new 404c022 [CARBONDATA-3244] Add benchmark for Change Data Capture
scenario
404c022 is described below
commit 404c02222be243cf3d1f225b01bd3642995f46ba
Author: Jacky Li <jacky.li...@qq.com>
AuthorDate: Fri Jan 11 00:44:58 2019 +0800
[CARBONDATA-3244] Add benchmark for Change Data Capture scenario
CDC (change data capture) is a common scenario for analyzing slowly changed
table in data warehouse.
It is good to add benchmark test comparing two update methods:
1. hive_solution, which uses INSERT OVERWRITE. This is a popular method for
hive warehouse.
2. carbon_solution, which uses CarbonData's update syntax to update the
history table directly.
This closes #3066
---
.../carbondata/benchmark/SCDType2Benchmark.scala | 300 +++++++++++++++++++++
1 file changed, 300 insertions(+)
diff --git
a/examples/spark2/src/main/scala/org/apache/carbondata/benchmark/SCDType2Benchmark.scala
b/examples/spark2/src/main/scala/org/apache/carbondata/benchmark/SCDType2Benchmark.scala
new file mode 100644
index 0000000..a2e1b6e
--- /dev/null
+++
b/examples/spark2/src/main/scala/org/apache/carbondata/benchmark/SCDType2Benchmark.scala
@@ -0,0 +1,300 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.benchmark
+
+import java.io.File
+import java.sql.Date
+
+import org.apache.commons.lang3.time.DateUtils
+import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
+
+/**
+ * Benchmark for SCD (Slowly Change Dimension) Type 2 performance.
+ * See [[https://en.wikipedia.org/wiki/Slowly_changing_dimension]]
+ *
+ * The benchmark shows performance of two update methods:
+ * 1. overwrite_solution, which uses INSERT OVERWRITE. This is a popular
method for hive warehouse.
+ * 2. update_solution, which uses CarbonData's update syntax to update the
history table directly.
+ *
+ * When running in a 8-cores laptop, the benchmark shows:
+ *
+ * 1. test one
+ * History table 1M records, update 10K records everyday and insert 10K
records everyday,
+ * simulated 3 days.
+ * hive_solution: total process time takes 13s
+ * carbon_solution: total process time takes 6s
+ *
+ *
+ * 2. test two
+ * History table 10M records, update 10K records everyday and insert 10K
records everyday,
+ * simulated 3 days.
+ * hive_solution: total process time takes 115s
+ * carbon_solution: total process time takes 15s
+ *
+ */
+// scalastyle:off println
+object SCDType2Benchmark {
+
+ // Schema for history table
+ // Table name: dw_order_solution1 (using overwrite), dw_order_solution2
(using update)
+ // +-------------+-----------+-------------+
+ // | Column name | Data type | Cardinality |
+ // +-------------+-----------+-------------+
+ // | order_id | string | 10,000,000 |
+ // +-------------+-----------+-------------+
+ // | customer_id | string | 10,000,000 |
+ // +-------------+-----------+-------------+
+ // | start_date | date | NA |
+ // +-------------+-----------+-------------+
+ // | end_date | date | NA |
+ // +-------------+-----------+-------------+
+ // | state | int | 4 |
+ // +-------------+-----------+-------------+
+ case class Order (order_id: String, customer_id: String, start_date: Date,
end_date: Date,
+ state: Int)
+
+ // Schema for Change table which is used for update to history table every
day
+ // Table name: ods_order
+ // +-------------+-----------+-------------+
+ // | Column name | Data type | Cardinality |
+ // +-------------+-----------+-------------+
+ // | order_id | string | 10,000,000 |
+ // +-------------+-----------+-------------+
+ // | customer_id | string | 10,000,000 |
+ // +-------------+-----------+-------------+
+ // | update_date | date | NA |
+ // +-------------+-----------+-------------+
+ // | state | int | 4 |
+ // +-------------+-----------+-------------+
+ case class Change (order_id: String, customer_id: String, update_date: Date,
state: Int)
+
+ // number of records for first day
+ val numOrders = 10000000
+
+ // number of records to update every day
+ val numUpdateOrdersDaily = 10000
+
+ // number of new records to insert every day
+ val newNewOrdersDaily = 10000
+
+ // number of days to simulate
+ val numDays = 3
+
+ // print eveyday result or not to console
+ val printDetail = false
+
+ def generateDataForDay0(
+ sparkSession: SparkSession,
+ numOrders: Int = 1000000,
+ startDate: Date = Date.valueOf("2018-05-01")): DataFrame = {
+ import sparkSession.implicits._
+ sparkSession.sparkContext.parallelize(1 to numOrders, 4)
+ .map { x => Order(s"order$x", s"customer$x", startDate,
Date.valueOf("9999-01-01"), 1)
+ }.toDS().toDF()
+ }
+
+ def generateDailyChange(
+ sparkSession: SparkSession,
+ numUpdatedOrders: Int,
+ startDate: Date,
+ updateDate: Date,
+ newState: Int,
+ numNewOrders: Int
+ ): DataFrame = {
+ import sparkSession.implicits._
+ // data for update to the history table
+ val ds1 = sparkSession.sparkContext.parallelize(1 to numUpdatedOrders, 4)
+ .map {x => Change(s"order$x", s"customer$x", updateDate, newState)
+ }.toDS().toDF()
+ // date for insert to the history table
+ val ds2 = sparkSession.sparkContext.parallelize(1 to numNewOrders, 4)
+ .map {x => Change(s"newOrder${System.currentTimeMillis()}",
s"customer$x", updateDate, 1)
+ }.toDS().toDF()
+ // union them so that the Change table contains both data for update and
insert
+ ds1.union(ds2)
+ }
+
+ def main(args: Array[String]): Unit = {
+ import org.apache.spark.sql.CarbonSession._
+ val rootPath = new File(this.getClass.getResource("/").getPath
+ + "../../../..").getCanonicalPath
+
+ val spark = SparkSession
+ .builder()
+ .master("local[8]")
+ .enableHiveSupport()
+ .config("spark.sql.warehouse.dir",
s"$rootPath/examples/spark2/target/warehouse")
+ .getOrCreateCarbonSession()
+ spark.sparkContext.setLogLevel("error")
+
+ // prepare base data for first day
+ spark.sql("drop table if exists dw_order_solution1")
+ spark.sql("drop table if exists dw_order_solution2")
+ spark.sql("drop table if exists change")
+
+ val baseData = generateDataForDay0(
+ sparkSession = spark,
+ numOrders = numOrders,
+ startDate = Date.valueOf("2018-05-01"))
+
+ baseData.write
+ .format("carbondata")
+ .option("tableName", "dw_order_solution1")
+ .mode(SaveMode.Overwrite)
+ .save()
+
+ baseData.write
+ .format("carbondata")
+ .option("tableName", "dw_order_solution2")
+ .option("sort_columns", "order_id")
+ .mode(SaveMode.Overwrite)
+ .save()
+
+ var startDate = Date.valueOf("2018-05-01")
+ var state = 2
+ var solution1UpdateTime = 0L
+ var solution2UpdateTime = 0L
+
+ if (printDetail) {
+ println("## day0")
+ spark.sql("select * from dw_order").show(100, false)
+ }
+
+ def timeIt(func: (SparkSession) => Unit): Long = {
+ val start = System.nanoTime()
+ func(spark)
+ System.nanoTime() - start
+ }
+
+ for (i <- 1 to numDays) {
+ // prepare for incremental update data for day-i
+ val newDate = new Date(DateUtils.addDays(startDate, 1).getTime)
+ val changeData = generateDailyChange(
+ sparkSession = spark,
+ numUpdatedOrders = numUpdateOrdersDaily,
+ startDate = startDate,
+ updateDate = newDate,
+ newState = state,
+ numNewOrders = newNewOrdersDaily)
+ changeData.write
+ .format("carbondata")
+ .option("tableName", "change")
+ .mode(SaveMode.Overwrite)
+ .save()
+
+ if (printDetail) {
+ println(s"day$i Change")
+ spark.sql("select * from change").show(100, false)
+ }
+
+ // apply Change to history table by using INSERT OVERWRITE
+ solution1UpdateTime += timeIt(solution1)
+
+ // apply Change to history table by using UPDATE
+ solution2UpdateTime += timeIt(solution2)
+
+ if (printDetail) {
+ println(s"day$i result")
+ spark.sql("select * from dw_order_solution1").show(false)
+ spark.sql("select * from dw_order_solution2").show(false)
+ }
+
+ startDate = newDate
+ state = state + 1
+ }
+
+ // do a query after apply SCD Type2 update
+ val solution1QueryTime = timeIt(
+ spark => spark.sql(
+ s"""
+ | select sum(state) as sum, customer_id
+ | from dw_order_solution1
+ | group by customer_id
+ | order by sum
+ | limit 10
+ |""".stripMargin).collect()
+ )
+
+ val solution2QueryTime = timeIt(
+ spark => spark.sql(
+ s"""
+ | select sum(state) as sum, customer_id
+ | from dw_order_solution2
+ | group by customer_id
+ | order by sum
+ | limit 10
+ |""".stripMargin).collect()
+ )
+
+ // print update time
+ println(s"overwrite solution update takes ${solution1UpdateTime / 1000 /
1000 / 1000} s")
+ println(s"update solution update takes ${solution2UpdateTime / 1000 / 1000
/ 1000} s")
+
+ // print query time
+ println(s"overwrite solution query takes ${solution1QueryTime / 1000 /
1000 / 1000} s")
+ println(s"update solution query takes ${solution2QueryTime / 1000 / 1000 /
1000} s")
+
+ spark.close()
+ }
+
+ /**
+ * Typical solution when using hive
+ * This solution uses INSERT OVERWRITE to rewrite the whole table every day
+ */
+ private def solution1(spark: SparkSession) = {
+ spark.sql(
+ """
+ | insert overwrite table dw_order_solution1
+ | select * from
+ | (
+ | select A.order_id, A.customer_id, A.start_date,
+ | case when A.end_date > B.update_date then B.update_date
+ | else A.end_date
+ | end as end_date,
+ | A.state
+ | from dw_order_solution1 A
+ | left join change B
+ | on A.order_id = B.order_id
+ | union all
+ | select B.order_id, B.customer_id, B.update_date,
date("9999-01-01"), B.state
+ | from change B
+ | ) T
+ """.stripMargin)
+ }
+
+ /**
+ * Solution leveraging carbon's update syntax
+ */
+ private def solution2(spark: SparkSession) = {
+ spark.sql(
+ """
+ | update dw_order_solution2 A
+ | set (A.end_date) =
+ | (select B.update_date
+ | from change B
+ | where A.order_id = B.order_id and A.end_date > B.update_date)
+ """.stripMargin).show
+ spark.sql(
+ """
+ | insert into dw_order_solution2
+ | select B.order_id, B.customer_id, B.update_date, date('9999-12-30'),
B.state
+ | from change B
+ """.stripMargin)
+ }
+}
+// scalastyle:on println
