Add Hudi connector (#257)

* Add Hudi connector * Fix missing save action * Specified Hudi spark version per profile * Update Hudi test case * Update Spark 2.3 Hudi version * Fix Hudi test with dedicated spark session * Update Hudi version for Spark 2.3 * Update work version for different spark profile 3.0 & 2.4 * Add HudiConnectorConf Tests * Add more test cases Co-authored-by: huong.vuong <huong.vuong@grabtaxi.com>

Author: hoaihuongbk

CONTRIBUTING.md

@@ -55,6 +55,15 @@ export SPARK_VER=2.4
 ./dev/test.sh
 ```
 
+Note: in some case you get the following error 
+```
+java.net.BindException: Can't assign requested address: Service 'sparkDriver'
+```
+then you have to bind the spark to local ip like this
+```shell
+export SPARK_LOCAL_IP=127.0.0.1
+```
+
 ## Styleguide
 
 ### Commit styleguide

pom.xml

@@ -98,6 +98,9 @@
  <maven.site.plugin.version>3.9.1</maven.site.plugin.version>
  <nexus.staging.maven.plugin.version>1.6.8</nexus.staging.maven.plugin.version>
  <typesafe.config.version>1.4.2</typesafe.config.version>
+ <hudi.version>0.11.0</hudi.version>
+ <hudi.bundle.version>spark3.2-bundle</hudi.bundle.version>
+ <spark.avro.version>3.0.2</spark.avro.version>
  </properties>
 
  <dependencies>
@@ -178,6 +181,18 @@
  <version>${delta.version}</version>
  </dependency>
 
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-${hudi.bundle.version}_${scala.compat.version}</artifactId>
+ <version>${hudi.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-avro_${scala.compat.version}</artifactId>
+ <version>${spark.avro.version}</version>
+ </dependency>
+
  <!-- TYPESAFE CONFIG -->
  <dependency>
  <groupId>com.typesafe</groupId>
@@ -393,6 +408,8 @@
  <spark.dynamodb.version>1.0.4</spark.dynamodb.version>
  <delta.version>0.6.1</delta.version>
  <apache.hadoop.version>2.9.2</apache.hadoop.version>
+ <hudi.bundle.version>spark2.4-bundle</hudi.bundle.version>
+ <spark.avro.version>2.4.8</spark.avro.version>
  </properties>
  </profile>
  <profile>
@@ -404,6 +421,9 @@
  <spark.dynamodb.version>1.0.4</spark.dynamodb.version>
  <delta.version>0.6.1</delta.version>
  <apache.hadoop.version>2.9.2</apache.hadoop.version>
+ <hudi.bundle.version>spark-bundle</hudi.bundle.version>
+ <hudi.version>0.7.0</hudi.version>
+ <spark.avro.version>2.4.4</spark.avro.version>
  </properties>
  </profile>
  <profile>
@@ -413,6 +433,8 @@
  <spark.compat.version>3.0</spark.compat.version>
  <spark.cassandra.connector.version>3.0.0</spark.cassandra.connector.version>
  <delta.version>0.7.0</delta.version>
+ <hudi.bundle.version>spark3.0.3-bundle</hudi.bundle.version>
+ <hudi.version>0.10.1</hudi.version>
  </properties>
  </profile>
  <profile>

src/main/java/io/github/setl/enums/Storage.java

@@ -13,6 +13,7 @@ public enum Storage {
  JSON("io.github.setl.storage.connector.JSONConnector"),
  JDBC("io.github.setl.storage.connector.JDBCConnector"),
  STRUCTURED_STREAMING("io.github.setl.storage.connector.StructuredStreamingConnector"),
+ HUDI("io.github.setl.storage.connector.HudiConnector"),
  OTHER(null);
 
  private String connectorName;

src/main/scala/io/github/setl/config/HudiConnectorConf.scala

@@ -0,0 +1,40 @@
+package io.github.setl.config
+
+import io.github.setl.exception.ConfException
+import org.apache.spark.sql.SaveMode
+
+class HudiConnectorConf extends ConnectorConf {
+
+ import HudiConnectorConf._
+
+ override def getReaderConf: Map[String, String] = {
+ import scala.collection.JavaConverters._
+ settings.asScala.toMap - PATH
+ }
+
+ override def getWriterConf: Map[String, String] = {
+ import scala.collection.JavaConverters._
+ settings.asScala.toMap - SAVEMODE - PATH
+ }
+
+ def setPath(path: String): this.type = set("path", path)
+
+ def setSaveMode(saveMode: String): this.type = set("saveMode", saveMode)
+
+ def setSaveMode(saveMode: SaveMode): this.type = set("saveMode", saveMode.toString)
+
+ def getPath: String = get("path") match {
+ case Some(path) => path
+ case _ => throw new ConfException("The value of path is not set")
+ }
+
+ def getSaveMode: SaveMode = SaveMode.valueOf(get("saveMode", SaveMode.Append.toString))
+
+}
+
+object HudiConnectorConf {
+ def fromMap(options: Map[String, String]): HudiConnectorConf = new HudiConnectorConf().set(options)
+
+ val SAVEMODE: String = "saveMode"
+ val PATH: String = "path"
+}

src/main/scala/io/github/setl/storage/connector/HudiConnector.scala

@@ -0,0 +1,66 @@
+package io.github.setl.storage.connector
+
+import com.typesafe.config.Config
+import io.github.setl.config.{Conf, HudiConnectorConf}
+import io.github.setl.enums.Storage
+import io.github.setl.internal.HasReaderWriter
+import io.github.setl.util.TypesafeConfigUtils
+import org.apache.spark.sql._
+
+class HudiConnector(val options: HudiConnectorConf) extends Connector with HasReaderWriter {
+ override val storage: Storage = Storage.HUDI
+
+ def this(options: Map[String, String]) = this(HudiConnectorConf.fromMap(options))
+
+ def this(config: Config) = this(TypesafeConfigUtils.getMap(config))
+
+ def this(conf: Conf) = this(conf.toMap)
+
+ override val reader: DataFrameReader = {
+ spark.read
+ .format("hudi")
+ .options(options.getReaderConf)
+ }
+
+ override val writer: DataFrame => DataFrameWriter[Row] = (df: DataFrame) => {
+ df.write
+ .format("hudi")
+ .mode(options.getSaveMode)
+ .options(options.getWriterConf)
+ }
+
+ /**
+ * Read data from the data source
+ *
+ * @return a [[DataFrame]]
+ */
+ @throws[java.io.FileNotFoundException](s"${options.getPath} doesn't exist")
+ @throws[org.apache.spark.sql.AnalysisException](s"${options.getPath} doesn't exist")
+ override def read(): DataFrame = {
+ logDebug(s"Reading ${storage.toString} file in: '${options.getPath}'")
+ this.setJobDescription(s"Read file(s) from '${options.getPath}'")
+ reader.load(options.getPath)
+ }
+
+ /**
+ * Write a [[DataFrame]] into the data storage
+ *
+ * @param t a [[DataFrame]] to be saved
+ * @param suffix for data connectors that support suffix (e.g. [[FileConnector]],
+ * add the given suffix to the save path
+ */
+ override def write(t: DataFrame, suffix: Option[String]): Unit = {
+ if (suffix.isDefined) logWarning("Suffix is not supported in HudiConnector")
+ write(t)
+ }
+
+ /**
+ * Write a [[DataFrame]] into the data storage
+ *
+ * @param t a [[DataFrame]] to be saved
+ */
+ override def write(t: DataFrame): Unit = {
+ this.setJobDescription(s"Write file to ${options.getPath}")
+ writer(t).save(options.getPath)
+ }
+}

src/test/resources/application.conf

@@ -252,3 +252,14 @@ schemaConverter {
  saveMode = "ErrorIfExists"
  }
 }
+
+hudi {
+ test {
+ path = "${project.basedir}/src/test/resources/test_hudi_7"
+ saveMode = "Overwrite"
+ hoodie.table.name = "test_object_7"
+ hoodie.datasource.write.recordkey.field = "col1"
+ hoodie.datasource.write.precombine.field = "col4"
+ hoodie.datasource.write.table.type = "MERGE_ON_READ"
+ }
+}

src/test/scala/io/github/setl/config/HudiConnectorConfSuite.scala

@@ -0,0 +1,48 @@
+package io.github.setl.config
+
+import io.github.setl.exception.ConfException
+import org.scalatest.funsuite.AnyFunSuite
+import org.apache.spark.sql.SaveMode
+
+class HudiConnectorConfSuite extends AnyFunSuite {
+ val conf = new HudiConnectorConf
+
+ test("Get/Set HudiConnectorConf") {
+ assert(conf.get("saveMode") === None)
+ conf.setSaveMode("Append")
+ assert(conf.getSaveMode === SaveMode.Append)
+ conf.setSaveMode("Overwrite")
+ assert(conf.getSaveMode === SaveMode.Overwrite)
+ conf.setSaveMode(SaveMode.Overwrite)
+ assert(conf.getSaveMode === SaveMode.Overwrite)
+
+ assert(conf.get("path") === None)
+ assertThrows[ConfException](conf.getPath)
+
+ conf.setPath("path")
+ assert(conf.getPath === "path")
+ }
+
+ test("Init HudiConnectorConf from options") {
+ val options : Map[String, String] = Map(
+ "path" -> "path",
+ "saveMode" -> "Append",
+ "hoodie.table.name" -> "test_object",
+ "hoodie.datasource.write.recordkey.field" -> "col1",
+ "hoodie.datasource.write.precombine.field" -> "col4",
+ "hoodie.datasource.write.table.type" -> "MERGE_ON_READ"
+ )
+
+ val confFromOpts: HudiConnectorConf = HudiConnectorConf.fromMap(options)
+ assert(confFromOpts.getPath === "path")
+ assert(confFromOpts.getSaveMode === SaveMode.Append)
+
+ val readerOpts = confFromOpts.getReaderConf
+ val writerOpts = confFromOpts.getWriterConf
+
+ // Config should not contains path & save mode
+ assert(!readerOpts.contains("path"))
+ assert(!writerOpts.contains("path"))
+ assert(!writerOpts.contains("saveMode"))
+ }
+}

src/test/scala/io/github/setl/config/Properties.scala

@@ -25,6 +25,8 @@ object Properties {
 
  val jdbcConfig: Config = cl.getConfig("psql.test")
 
+ val hudiConfig : Config = cl.getConfig("hudi.test")
+
  val excelConfigConnector: Config = cl.getConfig("connector.excel")
  val cassandraConfigConnector: Config = cl.getConfig("connector.cassandra")
  val csvConfigConnector: Config = cl.getConfig("connector.csv")

src/test/scala/io/github/setl/storage/connector/HudiConnectorSuite.scala

@@ -0,0 +1,62 @@
+package io.github.setl.storage.connector
+
+import io.github.setl.config.{Conf, HudiConnectorConf, Properties}
+import io.github.setl.{SparkSessionBuilder, SparkTestUtils, TestObject2}
+import org.apache.spark.sql.{SaveMode, SparkSession}
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.nio.file.Paths
+import java.sql.{Date, Timestamp}
+
+class HudiConnectorSuite extends AnyFunSuite {
+
+ val path: String = Paths.get("src", "test", "resources", "test_hudi").toFile.getAbsolutePath
+ val saveMode = SaveMode.Overwrite
+
+ val options: Map[String, String] = Map[String, String](
+ "path" -> path,
+ "saveMode" -> saveMode.toString,
+ "hoodie.table.name" -> "test_object",
+ "hoodie.datasource.write.recordkey.field" -> "col1",
+ "hoodie.datasource.write.precombine.field" -> "col4",
+ "hoodie.datasource.write.table.type" -> "MERGE_ON_READ"
+ )
+
+ val testTable: Seq[TestObject2] = Seq(
+ TestObject2("string", 5, 0.000000001685400132103450D, new Timestamp(1557153268000L), new Date(1557100800000L), 999999999999999999L),
+ TestObject2("string2", 5, 0.000000001685400132103450D, new Timestamp(1557153268000L), new Date(1557100800000L), 999999999999999999L),
+ TestObject2("string3", 5, 0.000000001685400132103450D, new Timestamp(1557153268000L), new Date(1557100800000L), 999999999999999999L)
+ )
+
+ test("Instantiation of constructors") {
+
+ // New spark session here since Hudi only supports KryoSerializer
+ val spark: SparkSession = new SparkSessionBuilder().setEnv("local")
+ .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
+ .build()
+ .get()
+ assume(SparkTestUtils.checkSparkVersion("2.4"))
+
+ import spark.implicits._
+
+ val connector = new HudiConnector(HudiConnectorConf.fromMap(options))
+ connector.write(testTable.toDF)
+ assert(connector.read().collect().length == testTable.length)
+
+ val path2: String = Paths.get("src", "test", "resources", "test_hudi_2").toFile.getAbsolutePath
+ val options2 = options + ("path" -> path2)
+ val connector2 = new HudiConnector(options2)
+ connector2.write(testTable.toDF)
+ assert(connector2.read().collect().length == testTable.length)
+
+ val path3: String = Paths.get("src", "test", "resources", "test_hudi_3").toFile.getAbsolutePath
+ val options3 = options + ("path" -> path3)
+ val connector3 = new HudiConnector(Conf.fromMap(options3))
+ connector3.write(testTable.toDF, Some("any_"))
+ assert(connector3.read().collect().length == testTable.length)
+
+ val connector7 = new HudiConnector(Properties.hudiConfig)
+ connector7.write(testTable.toDF)
+ assert(connector7.read().collect().length == testTable.length)
+ }
+}