Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[TEST] Add Date test for Spark on ClickHouse #345

Merged
merged 1 commit into from
May 21, 2021
Merged

[TEST] Add Date test for Spark on ClickHouse #345

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
104 changes: 104 additions & 0 deletions ...se-integration-spark/src/test/scala/com/github/housepower/jdbc/spark/SparkDateITest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,104 @@
/*
* Licensed 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 com.github.housepower.jdbc.spark

import com.github.housepower.annotation.Issue
import com.github.housepower.jdbc.AbstractITest
import com.github.housepower.jdbc.tool.TestHarness
import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.to_date
import org.apache.spark.sql.jdbc.{ClickHouseDialect, JdbcDialects}
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
import org.junit.jupiter.api.{BeforeAll, Test}

import java.sql.Date
import java.time.LocalDate
import scala.collection.Seq

object SparkDateITest {
@BeforeAll
def beforeAll(): Unit = {
// make sure register `ClickHouseDialects` before we use it
JdbcDialects.registerDialect(ClickHouseDialect)
}
}


class SparkDateITest extends AbstractITest with Logging {

@Test
@Issue(channel = "GitHub", value = "#343",
link = "https://github.com/housepower/ClickHouse-Native-JDBC/issues/343")
def testSparkJdbcWrite(): Unit = {
val helper = new TestHarness("Date")
helper.clean()
helper.create()
doSparkWriteAndRead(helper.getTableName)
helper.clean()
}

@transient lazy implicit val spark: SparkSession = {
SparkSession.builder()
.master("local[2]")
.appName("spark-ut")
.config("spark.ui.enabled", "false")
.config("spark.driver.host", "localhost")
.config("spark.sql.shuffle.partitions", "1")
.config("spark.sql.warehouse.dir", System.getProperty("java.io.tmpdir"))
.getOrCreate()
}

// col_0 Date
@transient lazy implicit val schema: StructType = StructType.apply(
StructField("col_0", DataTypes.DateType, nullable = false) :: Nil)

private def doSparkWriteAndRead(table: String): Unit = {
import spark.implicits._

val df = Seq("2020-10-27")
.toDF("col_0")
.withColumn("col_0", to_date($"col_0"))

val resultDf = spark.createDataFrame(df.rdd, schema)

resultDf
.write
.format("jdbc")
.mode("overwrite")
.option("driver", "com.github.housepower.jdbc.ClickHouseDriver")
.option("url", getJdbcUrl)
.option("user", "default")
.option("password", "")
.option("dbtable", table)
.option("truncate", "true")
.option("batchsize", 1000)
.option("isolationLevel", "NONE")
.save

val rows = spark.read
.format("jdbc")
.option("driver", "com.github.housepower.jdbc.ClickHouseDriver")
.option("url", getJdbcUrl)
.option("user", "default")
.option("password", "")
.option("dbtable", table)
.load
.collect

assert(rows.length == 1)
assert(rows(0).getDate(0) == Date.valueOf(LocalDate.of(2020, 10, 27)))
}
}
28 changes: 16 additions & 12 deletions ...ration-spark/src/test/scala/com/github/housepower/jdbc/spark/SparkOnClickHouseITest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ import com.github.housepower.jdbc.AbstractITest
import com.github.housepower.jdbc.tool.TestHarness
import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.to_timestamp
import org.apache.spark.sql.functions.{to_date, to_timestamp}
import org.apache.spark.sql.jdbc.{ClickHouseDialect, JdbcDialects}
import org.apache.spark.sql.types.{ArrayType, DataTypes, StructField, StructType}
import org.junit.jupiter.api.Assertions.assertEquals
Expand Down Expand Up @@ -102,8 +102,9 @@ class SparkOnClickHouseITest extends AbstractITest with Logging {
// col_9 Float64,
// col_10 Nullable(Float64),
// col_11 String,
// col_12 DateTime,
// col_13 Array(String)
// col_12 Date,
// col_13 DateTime,
// col_14 Array(String)
@transient lazy implicit val schema: StructType = StructType.apply(
StructField("col_0", DataTypes.ByteType, nullable = false) ::
StructField("col_1", DataTypes.ShortType, nullable = false) ::
Expand All @@ -117,17 +118,19 @@ class SparkOnClickHouseITest extends AbstractITest with Logging {
StructField("col_9", DataTypes.DoubleType, nullable = false) ::
StructField("col_10", DataTypes.DoubleType, nullable = true) ::
StructField("col_11", DataTypes.StringType, nullable = false) ::
StructField("col_12", DataTypes.TimestampType, nullable = false) ::
StructField("col_13", ArrayType(DataTypes.StringType, containsNull = false), nullable = false) :: Nil)
StructField("col_12", DataTypes.DateType, nullable = false) ::
StructField("col_13", DataTypes.TimestampType, nullable = false) ::
StructField("col_14", ArrayType(DataTypes.StringType, containsNull = false), nullable = false) :: Nil)

private def doSparkJdbcWrite(table: String): Unit = {
import spark.implicits._

val df = Seq((1.toByte, 1.toShort, 1, 1L, 1.toShort, 1, 1L, 1L, 1.1F, 1.1D, null, "a_1", "2020-10-27 01:46:45", Array("哈哈", "哇咔咔", "你好，世界")),
(2.toByte, 2.toShort, 2, 2L, 2.toShort, 2, 2L, 2L, 2.2F, 2.2D, null, "b_2", "2020-10-27 02:46:45", Array("🇨🇳", "🇷🇺", "🇩🇪", "🇯🇵", "🇺🇸")),
(3.toByte, 3.toShort, 3, 3L, 3.toShort, 3, 3L, 3L, 3.3F, 3.3D, null, "c_3", "2020-10-27 03:46:45", Array[String]()))
.toDF("col_0", "col_1", "col_2", "col_3", "col_4", "col_5", "col_6", "col_7", "col_8", "col_9", "col_10", "col_11", "col_12", "col_13")
.withColumn("col_12", to_timestamp($"col_12"))
val df = Seq((1.toByte, 1.toShort, 1, 1L, 1.toShort, 1, 1L, 1L, 1.1F, 1.1D, null, "a_1", "2020-10-27", "2020-10-27 01:46:45", Array("哈哈", "哇咔咔", "你好，世界")),
(2.toByte, 2.toShort, 2, 2L, 2.toShort, 2, 2L, 2L, 2.2F, 2.2D, null, "b_2", "2020-10-27", "2020-10-27 02:46:45", Array("🇨🇳", "🇷🇺", "🇩🇪", "🇯🇵", "🇺🇸")),
(3.toByte, 3.toShort, 3, 3L, 3.toShort, 3, 3L, 3L, 3.3F, 3.3D, null, "c_3", "2020-10-27", "2020-10-27 03:46:45", Array[String]()))
.toDF("col_0", "col_1", "col_2", "col_3", "col_4", "col_5", "col_6", "col_7", "col_8", "col_9", "col_10", "col_11", "col_12", "col_13", "col_14")
.withColumn("col_12", to_date($"col_12"))
.withColumn("col_13", to_timestamp($"col_13"))

val resultDf = spark.createDataFrame(df.rdd, schema)

Expand Down Expand Up @@ -192,8 +195,9 @@ class SparkOnClickHouseITest extends AbstractITest with Logging {
StructField("col_9", DataTypes.DoubleType, nullable = true) ::
StructField("col_10", DataTypes.DoubleType, nullable = true) ::
StructField("col_11", DataTypes.StringType, nullable = true) ::
StructField("col_12", DataTypes.TimestampType, nullable = true) ::
StructField("col_13", ArrayType(DataTypes.StringType, containsNull = false), nullable = true) :: Nil),
StructField("col_12", DataTypes.DateType, nullable = true) ::
StructField("col_13", DataTypes.TimestampType, nullable = true) ::
StructField("col_14", ArrayType(DataTypes.StringType, containsNull = false), nullable = true) :: Nil),
df.schema)

df.write
Expand Down
Loading