[HUDI-1415] Read Hoodie Table As Spark DataSource Table #2283
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -36,6 +36,7 @@ import org.apache.hudi.common.util.{CommitUtils, ReflectionUtils} | |
| import org.apache.hudi.config.HoodieBootstrapConfig.{BOOTSTRAP_BASE_PATH_PROP, BOOTSTRAP_INDEX_CLASS_PROP, DEFAULT_BOOTSTRAP_INDEX_CLASS} | ||
| import org.apache.hudi.config.HoodieWriteConfig | ||
| import org.apache.hudi.exception.HoodieException | ||
| import org.apache.hudi.hive.util.ConfigUtils | ||
| import org.apache.hudi.hive.{HiveSyncConfig, HiveSyncTool} | ||
| import org.apache.hudi.internal.DataSourceInternalWriterHelper | ||
| import org.apache.hudi.sync.common.AbstractSyncTool | ||
|
|
@@ -44,7 +45,10 @@ import org.apache.spark.SPARK_VERSION | |
| import org.apache.spark.SparkContext | ||
| import org.apache.spark.api.java.JavaSparkContext | ||
| import org.apache.spark.rdd.RDD | ||
| import org.apache.spark.sql.internal.SQLConf | ||
| import org.apache.spark.sql.internal.StaticSQLConf.SCHEMA_STRING_LENGTH_THRESHOLD | ||
| import org.apache.spark.sql.types.StructType | ||
| import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode, SparkSession} | ||
|
|
||
| import scala.collection.JavaConversions._ | ||
| import scala.collection.mutable.ListBuffer | ||
|
|
@@ -220,7 +224,8 @@ private[hudi] object HoodieSparkSqlWriter { | |
|
|
||
| // Check for errors and commit the write. | ||
| val (writeSuccessful, compactionInstant) = | ||
| commitAndPerformPostOperations(sqlContext.sparkSession, df.schema, | ||
| writeResult, parameters, writeClient, tableConfig, jsc, | ||
| TableInstantInfo(basePath, instantTime, commitActionType, operation)) | ||
|
|
||
| def unpersistRdd(rdd: RDD[_]): Unit = { | ||
|
|
@@ -305,7 +310,7 @@ private[hudi] object HoodieSparkSqlWriter { | |
| } finally { | ||
| writeClient.close() | ||
| } | ||
| val metaSyncSuccess = metaSync(sqlContext.sparkSession, parameters, basePath, df.schema) | ||
| metaSyncSuccess | ||
| } | ||
|
|
||
|
|
@@ -346,12 +351,13 @@ private[hudi] object HoodieSparkSqlWriter { | |
| } | ||
| val hiveSyncEnabled = params.get(HIVE_SYNC_ENABLED_OPT_KEY).exists(r => r.toBoolean) | ||
| val metaSyncEnabled = params.get(META_SYNC_ENABLED_OPT_KEY).exists(r => r.toBoolean) | ||
| val syncHiveSuccess = | ||
| if (hiveSyncEnabled || metaSyncEnabled) { | ||
| metaSync(sqlContext.sparkSession, parameters, basePath, df.schema) | ||
| } else { | ||
| true | ||
| } | ||
| (syncHiveSuccess, common.util.Option.ofNullable(instantTime)) | ||
| } | ||
|
|
||
| def toProperties(params: Map[String, String]): TypedProperties = { | ||
|
|
@@ -398,7 +404,7 @@ private[hudi] object HoodieSparkSqlWriter { | |
| private def buildSyncConfig(basePath: Path, parameters: Map[String, String]): HiveSyncConfig = { | ||
| val hiveSyncConfig: HiveSyncConfig = new HiveSyncConfig() | ||
| hiveSyncConfig.basePath = basePath.toString | ||
| hiveSyncConfig.baseFileFormat = parameters(HIVE_BASE_FILE_FORMAT_OPT_KEY) | ||
| hiveSyncConfig.usePreApacheInputFormat = | ||
| parameters.get(HIVE_USE_PRE_APACHE_INPUT_FORMAT_OPT_KEY).exists(r => r.toBoolean) | ||
| hiveSyncConfig.databaseName = parameters(HIVE_DATABASE_OPT_KEY) | ||
|
|
@@ -417,35 +423,95 @@ private[hudi] object HoodieSparkSqlWriter { | |
| hiveSyncConfig.autoCreateDatabase = parameters.get(HIVE_AUTO_CREATE_DATABASE_OPT_KEY).exists(r => r.toBoolean) | ||
| hiveSyncConfig.decodePartition = parameters.getOrElse(URL_ENCODE_PARTITIONING_OPT_KEY, | ||
| DEFAULT_URL_ENCODE_PARTITIONING_OPT_VAL).toBoolean | ||
|
|
||
| val syncAsDtaSourceTable = parameters.getOrElse(DataSourceWriteOptions.HIVE_SYNC_AS_DATA_SOURCE_TABLE, | ||
| DataSourceWriteOptions.DEFAULT_HIVE_SYNC_AS_DATA_SOURCE_TABLE).toBoolean | ||
| if (syncAsDtaSourceTable) { | ||
| hiveSyncConfig.tableProperties = parameters.getOrElse(HIVE_TABLE_PROPERTIES, null) | ||
| hiveSyncConfig.serdeProperties = createSqlTableSerdeProperties(parameters, basePath.toString) | ||
| } | ||
| hiveSyncConfig | ||
| } | ||
|
|
||
| /** | ||
| * Add Spark Sql related table properties to the HIVE_TABLE_PROPERTIES. | ||
| * @param sqlConf The spark sql conf. | ||
| * @param schema The schema to write to the table. | ||
| * @param parameters The origin parameters. | ||
| * @return A new parameters added the HIVE_TABLE_PROPERTIES property. | ||
| */ | ||
| private def addSqlTableProperties(sqlConf: SQLConf, schema: StructType, | ||
| parameters: Map[String, String]): Map[String, String] = { | ||
| // Convert the schema and partition info used by spark sql to hive table properties. | ||
| // The following code refers to the spark code in | ||
| // https://github.com/apache/spark/blob/master/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala | ||
|
|
||
| val partitionSet = parameters(HIVE_PARTITION_FIELDS_OPT_KEY) | ||
| .split(",").map(_.trim).filter(!_.isEmpty).toSet | ||
| val threshold = sqlConf.getConf(SCHEMA_STRING_LENGTH_THRESHOLD) | ||
|
|
||
| val (partitionCols, dataCols) = schema.partition(c => partitionSet.contains(c.name)) | ||
| val reOrderedType = StructType(dataCols ++ partitionCols) | ||
| val schemaParts = reOrderedType.json.grouped(threshold).toSeq | ||
|
|
||
| var properties = Map( | ||
| "spark.sql.sources.provider" -> "hudi", | ||
umehrot2 marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| "spark.sql.sources.schema.numParts" -> schemaParts.size.toString | ||
|
||
| ) | ||
| schemaParts.zipWithIndex.foreach { case (part, index) => | ||
| properties += s"spark.sql.sources.schema.part.$index" -> part | ||
| } | ||
| // add partition columns | ||
| if (partitionSet.nonEmpty) { | ||
| properties += "spark.sql.sources.schema.numPartCols" -> partitionSet.size.toString | ||
| partitionSet.zipWithIndex.foreach { case (partCol, index) => | ||
| properties += s"spark.sql.sources.schema.partCol.$index" -> partCol | ||
| } | ||
| } | ||
| var sqlPropertyText = ConfigUtils.configToString(properties) | ||
| sqlPropertyText = if (parameters.containsKey(HIVE_TABLE_PROPERTIES)) { | ||
| sqlPropertyText + "\n" + parameters(HIVE_TABLE_PROPERTIES) | ||
| } else { | ||
| sqlPropertyText | ||
| } | ||
| parameters + (HIVE_TABLE_PROPERTIES -> sqlPropertyText) | ||
| } | ||
|
|
||
| private def createSqlTableSerdeProperties(parameters: Map[String, String], basePath: String): String = { | ||
| val pathProp = s"path=$basePath" | ||
| if (parameters.containsKey(HIVE_TABLE_SERDE_PROPERTIES)) { | ||
| pathProp + "\n" + parameters(HIVE_TABLE_SERDE_PROPERTIES) | ||
| } else { | ||
| pathProp | ||
| } | ||
| } | ||
|
|
||
| private def metaSync(spark: SparkSession, parameters: Map[String, String], basePath: Path, | ||
| schema: StructType): Boolean = { | ||
| val hiveSyncEnabled = parameters.get(HIVE_SYNC_ENABLED_OPT_KEY).exists(r => r.toBoolean) | ||
| var metaSyncEnabled = parameters.get(META_SYNC_ENABLED_OPT_KEY).exists(r => r.toBoolean) | ||
| var syncClientToolClassSet = scala.collection.mutable.Set[String]() | ||
| parameters(META_SYNC_CLIENT_TOOL_CLASS).split(",").foreach(syncClass => syncClientToolClassSet += syncClass) | ||
|
|
||
| val newParameters = addSqlTableProperties(spark.sessionState.conf, schema, parameters) | ||
| // for backward compatibility | ||
| if (hiveSyncEnabled) { | ||
| metaSyncEnabled = true | ||
| syncClientToolClassSet += classOf[HiveSyncTool].getName | ||
| } | ||
| var metaSyncSuccess = true | ||
| if (metaSyncEnabled) { | ||
| val fs = basePath.getFileSystem(spark.sessionState.newHadoopConf()) | ||
| syncClientToolClassSet.foreach(impl => { | ||
| val syncSuccess = impl.trim match { | ||
| case "org.apache.hudi.hive.HiveSyncTool" => { | ||
| log.info("Syncing to Hive Metastore (URL: " + parameters(HIVE_URL_OPT_KEY) + ")") | ||
| syncHive(basePath, fs, newParameters) | ||
| true | ||
| } | ||
| case _ => { | ||
| val properties = new Properties(); | ||
| properties.putAll(newParameters) | ||
| properties.put("basePath", basePath.toString) | ||
| val syncHoodie = ReflectionUtils.loadClass(impl.trim, Array[Class[_]](classOf[Properties], classOf[FileSystem]), properties, fs).asInstanceOf[AbstractSyncTool] | ||
| syncHoodie.syncHoodieTable() | ||
|
|
@@ -463,7 +529,9 @@ private[hudi] object HoodieSparkSqlWriter { | |
| */ | ||
| case class TableInstantInfo(basePath: Path, instantTime: String, commitActionType: String, operation: WriteOperationType) | ||
|
|
||
| private def commitAndPerformPostOperations(spark: SparkSession, | ||
| schema: StructType, | ||
| writeResult: HoodieWriteResult, | ||
| parameters: Map[String, String], | ||
| client: SparkRDDWriteClient[HoodieRecordPayload[Nothing]], | ||
| tableConfig: HoodieTableConfig, | ||
|
|
@@ -497,7 +565,8 @@ private[hudi] object HoodieSparkSqlWriter { | |
| } | ||
|
|
||
| log.info(s"Compaction Scheduled is $compactionInstant") | ||
|
|
||
| val metaSyncSuccess = metaSync(spark, parameters, tableInstantInfo.basePath, schema) | ||
|
|
||
| log.info(s"Is Async Compaction Enabled ? $asyncCompactionEnabled") | ||
| if (!asyncCompactionEnabled) { | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Lets introduce another additional boolean property
hoodie.datasource.hive_sync.sync_as_datasourceand put the feature behind it. We can usetrueby default, but atleast there would be a way to turn it off. This is going to change the way spark sql queries currently run with Hudi, and is a huge change.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good suggestion! +1 for this!