lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-1064]Trim hoodie table name (#1805)

Browse Source
This commit is contained in:
mabin001
2020-07-07 19:10:16 +08:00
committed by GitHub
parent be85a6c32b
commit 8c4ff185f1
1 changed files with 11 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
hudi-spark/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
View File

@@ -52,10 +52,11 @@ private[hudi] object HoodieSparkSqlWriter {
val sparkContext = sqlContext.sparkContext
val path = parameters.get("path")
val tblName = parameters.get(HoodieWriteConfig.TABLE_NAME)
if (path.isEmpty || tblName.isEmpty) {
val tblNameOp = parameters.get(HoodieWriteConfig.TABLE_NAME)
if (path.isEmpty || tblNameOp.isEmpty) {
throw new HoodieException(s"'${HoodieWriteConfig.TABLE_NAME}', 'path' must be set.")
}
val tblName = tblNameOp.get.trim
sparkContext.getConf.getOption("spark.serializer") match {
case Some(ser) if ser.equals("org.apache.spark.serializer.KryoSerializer") =>
case _ => throw new HoodieException("hoodie only support org.apache.spark.serializer.KryoSerializer as spark.serializer")
@@ -85,7 +86,7 @@ private[hudi] object HoodieSparkSqlWriter {
if (exists && mode == SaveMode.Append) {
val existingTableName = new HoodieTableMetaClient(sparkContext.hadoopConfiguration, path.get).getTableConfig.getTableName
if (!existingTableName.equals(tblName.get)) {
if (!existingTableName.equals(tblName)) {
throw new HoodieException(s"hoodie table with name $existingTableName already exist at $basePath")
}
}
@@ -93,8 +94,8 @@ private[hudi] object HoodieSparkSqlWriter {
val (writeStatuses, writeClient: HoodieWriteClient[HoodieRecordPayload[Nothing]]) =
if (!operation.equalsIgnoreCase(DELETE_OPERATION_OPT_VAL)) {
// register classes & schemas
val structName = s"${tblName.get}_record"
val nameSpace = s"hoodie.${tblName.get}"
val structName = s"${tblName}_record"
val nameSpace = s"hoodie.${tblName}"
sparkContext.getConf.registerKryoClasses(
Array(classOf[org.apache.avro.generic.GenericData],
classOf[org.apache.avro.Schema]))
@@ -129,11 +130,11 @@ private[hudi] object HoodieSparkSqlWriter {
// Create the table if not present
if (!exists) {
HoodieTableMetaClient.initTableType(sparkContext.hadoopConfiguration, path.get, tableType,
tblName.get, "archived", parameters(PAYLOAD_CLASS_OPT_KEY))
tblName, "archived", parameters(PAYLOAD_CLASS_OPT_KEY))
}
// Create a HoodieWriteClient & issue the write.
val client = DataSourceUtils.createHoodieClient(jsc, schema.toString, path.get, tblName.get,
val client = DataSourceUtils.createHoodieClient(jsc, schema.toString, path.get, tblName,
mapAsJavaMap(parameters)
)
@@ -158,8 +159,8 @@ private[hudi] object HoodieSparkSqlWriter {
throw new HoodieException(s"Append is the only save mode applicable for $operation operation")
}
val structName = s"${tblName.get}_record"
val nameSpace = s"hoodie.${tblName.get}"
val structName = s"${tblName}_record"
val nameSpace = s"hoodie.${tblName}"
sparkContext.getConf.registerKryoClasses(
Array(classOf[org.apache.avro.generic.GenericData],
classOf[org.apache.avro.Schema]))
@@ -175,7 +176,7 @@ private[hudi] object HoodieSparkSqlWriter {
// Create a HoodieWriteClient & issue the delete.
val client = DataSourceUtils.createHoodieClient(jsc,
Schema.create(Schema.Type.NULL).toString, path.get, tblName.get,
Schema.create(Schema.Type.NULL).toString, path.get, tblName,
mapAsJavaMap(parameters)
)