[SUPPORT] Does latest versions of Hudi (0.7.0, 0.6.0) work with Spark 2.3.0 when reading orc files?

[codejoyan]

Description:
The documentation says: Hudi works with Spark-2.x & Spark 3.x versions. (https://hudi.apache.org/docs/quick-start-guide.html) But I have not been able to use hudi-spark-bundle_2.11 version 0.7.0 with Spark 2.3.0 and Scala 2.11.12. Is there any specific spark_avro package one has to use?

I am trying to write a dataframe read from orc file
The job fails with the below error: java.lang.NoSuchMethodError: org.apache.spark.sql.types.Decimal$.minBytesForPrecision()[I Any inputs will be very helpful.

In the cluster I am working with we have Spark 2.3.0 and there is no immediate upgrade planned. Wanted to check if there is there any way to make Hudi 0.7.0 work with Spark 2.3.0?

Note: I am able to use Spark 2.3.0 with hudi-spark-bundle-0.5.0-incubating.jar

spark-shell Command
spark-shell
--packages org.apache.hudi:hudi-spark-bundle_2.11:0.7.0,org.apache.spark:spark-avro_2.11:2.4.4,org.apache.avro:avro:1.8.2
--conf spark.driver.extraClassPath=/u/users/j0s0j7j/.ivy2/jars/org.apache.avro_avro-1.8.2.jar
--conf spark.executor.extraClassPath=/u/users/j0s0j7j/.ivy2/jars/org.apache.avro_avro-1.8.2.jar
--conf "spark.sql.hive.convertMetastoreParquet=false"
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer'

Stack Strace
scala> val BaseTableIncrLoad = spark.sql(
s"""select * from $srcSchema.$srcTable
|where load_time >= ('2021-01-20') and load_time < ('2021-01-21')

val transformedDF = BaseTableIncrLoad.withColumn("partitionpath", concat(lit("part_str_col="), col("part_str_col"), lit("/part_date_col="), col("part_date_col")))

scala> transformedDF.write.format("org.apache.hudi").
| options(getQuickstartWriteConfigs).
| option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "col1").
| //option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "col2").
| option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "col3,col4,col5").
| option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
| option(DataSourceWriteOptions.KEYGENERATOR_CLASS_OPT_KEY, "org.apache.hudi.ComplexKeyGenerator").
| option("hoodie.upsert.shuffle.parallelism","20").
| option("hoodie.insert.shuffle.parallelism","20").
| option(HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT_BYTES, 128 * 1024 * 1024).
| option(HoodieStorageConfig.PARQUET_FILE_MAX_BYTES, 128 * 1024 * 1024).
| option(HoodieWriteConfig.TABLE_NAME, "targetTableHudi").
| mode(SaveMode.Append).
| save(targetPath)
21/02/22 07:14:03 WARN Utils: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.debug.maxToStringFields' in SparkEnv.conf.
java.lang.NoSuchMethodError: org.apache.spark.sql.types.Decimal$.minBytesForPrecision()[I
at org.apache.hudi.spark.org.apache.spark.sql.avro.SchemaConverters$.toAvroType(SchemaConverters.scala:156)
at org.apache.hudi.spark.org.apache.spark.sql.avro.SchemaConverters$$anonfun$5.apply(SchemaConverters.scala:176)
at org.apache.hudi.spark.org.apache.spark.sql.avro.SchemaConverters$$anonfun$5.apply(SchemaConverters.scala:174)
at scala.collection.Iterator$class.foreach(Iterator.scala:893)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at org.apache.spark.sql.types.StructType.foreach(StructType.scala:99)
at org.apache.hudi.spark.org.apache.spark.sql.avro.SchemaConverters$.toAvroType(SchemaConverters.scala:174)
at org.apache.hudi.AvroConversionUtils$.convertStructTypeToAvroSchema(AvroConversionUtils.scala:52)
at org.apache.hudi.HoodieSparkSqlWriter$.write(HoodieSparkSqlWriter.scala:139)
at org.apache.hudi.DefaultSource.createRelation(DefaultSource.scala:134)
at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:46)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:70)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:68)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:86)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:80)
at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:80)
at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:654)
at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:654)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:77)
at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:654)
at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:273)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:267)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:225)
... 62 elided

[bvaradar]

@codejoyan : Would it be possible for you to give us a reproducible set of steps (just like the Quickstart : https://hudi.apache.org/docs/quick-start-guide.html ) ?

[codejoyan]

Sure @bvaradar . Below steps will help to reproduce.
Observation: This is failing when the dataframe has specific datatypes decimal with precision. If it is other datatypes it succeeds

Environment Details:
Spark version in HDP - 2.3.0.2.6.5.279-2
Hudi: hudi-spark-bundle_2.11:0.7.0

Step-1: Create dummy data

$ hdfs dfs -cat hdfs://XXXXXw00ha/user/joyan/test_data.csv
col_1,col_2,col_3,col_4,col_5,col_6,col_7,col_8,col_9,col_10,col_11,col_12,cntry_cd,bus_dt
7IN00716079317820210109153408,716,3,AB,INR,107667253,0,1,2021-02-14 20:23:54.753,useridjsb91,2021-02-14 20:23:54.753,useridjsb91,IN,2021-02-01
7IN00716079317820210109153408,716,2,AB,INR,212733302,0,1,2021-02-14 20:23:54.753,useridjsb91,2021-02-14 20:23:54.753,useridjsb91,IN,2021-02-01
7IN00716079317820210109153408,716,1,AB,INR,139224013,0,1,2021-02-14 20:23:54.753,useridjsb91,2021-02-14 20:23:54.753,useridjsb91,IN,2021-02-01
AU700716079381819643325112243,5700,5,AB,INR,136158881,0,1,2021-02-14 20:06:38.718,useridjsb91,2021-02-14 20:06:38.718,useridjsb91,IN,2021-02-02
AU700716079381819643325112243,5700,6,AB,INR,585710940,4,1.97,2021-02-14 20:06:38.718,useridjsb91,2021-02-14 20:06:38.718,useridjsb91,IN,2021-02-02
AU700716079381819643325112243,5700,4,AB,INR,136158881,0,1,2021-02-14 20:06:38.718,useridjsb91,2021-02-14 20:06:38.718,useridjsb91,AU,2021-02-01
AU700716079381819643325112243,5700,3,AB,INR,136158881,0,1,2021-02-14 20:06:38.718,useridjsb91,2021-02-14 20:06:38.718,useridjsb91,AU,2021-02-01
AU700716079381819643325112243,5700,1,AB,INR,136158881,0,1,2021-02-14 20:06:38.718,useridjsb91,2021-02-14 20:06:38.718,useridjsb91,AU,2021-02-01

Step-2: Try to read the data from file, cast to appropriate datatypes and save as Hudi table

$ spark-shell \
--packages org.apache.hudi:hudi-spark-bundle_2.11:0.7.0,org.apache.spark:spark-avro_2.11:2.4.4,org.apache.avro:avro:1.8.2 \
--conf spark.driver.extraClassPath=/path/org.apache.avro_avro-1.8.2.jar \
--conf spark.executor.extraClassPath=/path/org.apache.avro_avro-1.8.2.jar \
--conf "spark.sql.hive.convertMetastoreParquet=false" \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer'

import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.QuickstartUtils.getQuickstartWriteConfigs
import org.apache.hudi.config.{HoodieCompactionConfig, HoodieStorageConfig, HoodieWriteConfig}
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.functions.{col, concat, lit}

val inputDF = spark.read.format("csv").option("header", "true").load("hdfs://XXXXXw00ha/user/joyan/test_data.csv")

val formattedDF = inputDF.selectExpr("col_1", "cast(col_2 as integer) col_2",
	"cast(col_3 as short) col_3", "col_4", "col_5", "cast(col_6 as byte) col_6", "cast(col_7 as decimal(9,2)) col_7",
	"cast(col_8 as decimal(9,2)) col_8", "cast(col_9 as timestamp) col_9", "col_10", "cast(col_11 as timestamp) col_11",
	"col_12", "cntry_cd", "cast(bus_dt as date) bus_dt")

formattedDF.printSchema
root
 |-- col_1: string (nullable = true)
 |-- col_2: integer (nullable = true)
 |-- col_3: short (nullable = true)
 |-- col_4: string (nullable = true)
 |-- col_5: string (nullable = true)
 |-- col_6: byte (nullable = true)
 |-- col_7: decimal(9,2) (nullable = true)
 |-- col_8: decimal(9,2) (nullable = true)
 |-- col_9: timestamp (nullable = true)
 |-- col_10: string (nullable = true)
 |-- col_11: timestamp (nullable = true)
 |-- col_12: string (nullable = true)
 |-- cntry_cd: string (nullable = true)
 |-- bus_dt: date (nullable = true)


formattedDF.show
+--------------------+-----+-----+-----+-----+-----+--------------------+-----------+--------------------+-----------+--------+----------+
|               col_1|col_2|col_3|col_4|col_5|col_6|               col_9|     col_10|              col_11|     col_12|cntry_cd|    bus_dt|
+--------------------+-----+-----+-----+-----+-----+--------------------+-----------+--------------------+-----------+--------+----------+
|7IN00716079317820...|  716|    3|   AB|  INR| null|2021-02-14 20:23:...|useridjsb91|2021-02-14 20:23:...|useridjsb91|      IN|2021-02-01|
|7IN00716079317820...|  716|    2|   AB|  INR| null|2021-02-14 20:23:...|useridjsb91|2021-02-14 20:23:...|useridjsb91|      IN|2021-02-01|
|7IN00716079317820...|  716|    1|   AB|  INR| null|2021-02-14 20:23:...|useridjsb91|2021-02-14 20:23:...|useridjsb91|      IN|2021-02-01|
|AU700716079381819...| 5700|    5|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      IN|2021-02-02|
|AU700716079381819...| 5700|    6|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      IN|2021-02-02|
|AU700716079381819...| 5700|    4|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      AU|2021-02-01|
|AU700716079381819...| 5700|    3|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      AU|2021-02-01|
|AU700716079381819...| 5700|    1|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      AU|2021-02-01|
+--------------------+-----+-----+-----+-----+-----+--------------------+-----------+--------------------+-----------+--------+----------+

val transformedDF = formattedDF.withColumn("partitionpath", concat(lit("cntry_cd="), col("cntry_cd"), lit("/bus_dt="), col("bus_dt")))

val targetPath = "gs://XXXXXXXXXXX7bb3d68/test_table_tgt"

transformedDF.write.format("org.apache.hudi").
options(getQuickstartWriteConfigs).
option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "col_9").
option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "col_2,col_1,col_3").
option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
option(DataSourceWriteOptions.KEYGENERATOR_CLASS_OPT_KEY, "org.apache.hudi.keygen.ComplexKeyGenerator").
option("hoodie.upsert.shuffle.parallelism","2").
option("hoodie.insert.shuffle.parallelism","2").
option(HoodieWriteConfig.TABLE_NAME, "targetTableHudi").
mode(SaveMode.Append).
save(targetPath)

java.lang.NoSuchMethodError: org.apache.spark.sql.types.Decimal$.minBytesForPrecision()[I
  at org.apache.hudi.spark.org.apache.spark.sql.avro.SchemaConverters$.toAvroType(SchemaConverters.scala:156)
  at org.apache.hudi.spark.org.apache.spark.sql.avro.SchemaConverters$$anonfun$5.apply(SchemaConverters.scala:176)
  at org.apache.hudi.spark.org.apache.spark.sql.avro.SchemaConverters$$anonfun$5.apply(SchemaConverters.scala:174)
  at scala.collection.Iterator$class.foreach(Iterator.scala:893)
  at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
  at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
  at org.apache.spark.sql.types.StructType.foreach(StructType.scala:99)
  at org.apache.hudi.spark.org.apache.spark.sql.avro.SchemaConverters$.toAvroType(SchemaConverters.scala:174)
  at org.apache.hudi.AvroConversionUtils$.convertStructTypeToAvroSchema(AvroConversionUtils.scala:52)
  at org.apache.hudi.HoodieSparkSqlWriter$.write(HoodieSparkSqlWriter.scala:139)
  at org.apache.hudi.DefaultSource.createRelation(DefaultSource.scala:134)
  at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:46)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:70)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:68)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:86)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
  at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
  at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:80)
  at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:80)
  at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:654)
  at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:654)
  at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:77)
  at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:654)
  at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:273)
  at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:267)
  at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:225)
  ... 59 elided

[codejoyan]

The below succeeds. I removed the decimal fields only and was able to save

scala> val inputDF = spark.read.format("csv").option("header", "true").load("hdfs://XXXXXhw00ha/user/joyan/test_data.csv")
inputDF: org.apache.spark.sql.DataFrame = [col_1: string, col_2: string ... 12 more fields]

scala> val formattedDF = inputDF.selectExpr("col_1", "cast(col_2 as integer) col_2",
     | "cast(col_3 as short) col_3", "col_4", "col_5", "cast(col_6 as byte) col_6",
     | "cast(col_9 as timestamp) col_9", "col_10", "cast(col_11 as timestamp) col_11",
     | "col_12", "cntry_cd", "cast(bus_dt as date) bus_dt")
formattedDF: org.apache.spark.sql.DataFrame = [col_1: string, col_2: int ... 10 more fields]

scala> val transformedDF = formattedDF.withColumn("partitionpath", concat(lit("cntry_cd="), col("cntry_cd"), lit("/bus_dt="), col("bus_dt")))
transformedDF: org.apache.spark.sql.DataFrame = [col_1: string, col_2: int ... 11 more fields]

scala> transformedDF.show
+--------------------+-----+-----+-----+-----+-----+--------------------+-----------+--------------------+-----------+--------+----------+--------------------+
|               col_1|col_2|col_3|col_4|col_5|col_6|               col_9|     col_10|              col_11|     col_12|cntry_cd|    bus_dt|       partitionpath|
+--------------------+-----+-----+-----+-----+-----+--------------------+-----------+--------------------+-----------+--------+----------+--------------------+
|7IN00716079317820...|  716|    3|   AB|  INR| null|2021-02-14 20:23:...|useridjsb91|2021-02-14 20:23:...|useridjsb91|      IN|2021-02-01|cntry_cd=IN/bus_d...|
|7IN00716079317820...|  716|    2|   AB|  INR| null|2021-02-14 20:23:...|useridjsb91|2021-02-14 20:23:...|useridjsb91|      IN|2021-02-01|cntry_cd=IN/bus_d...|
|7IN00716079317820...|  716|    1|   AB|  INR| null|2021-02-14 20:23:...|useridjsb91|2021-02-14 20:23:...|useridjsb91|      IN|2021-02-01|cntry_cd=IN/bus_d...|
|AU700716079381819...| 5700|    5|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      IN|2021-02-02|cntry_cd=IN/bus_d...|
|AU700716079381819...| 5700|    6|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      IN|2021-02-02|cntry_cd=IN/bus_d...|
|AU700716079381819...| 5700|    4|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      AU|2021-02-01|cntry_cd=AU/bus_d...|
|AU700716079381819...| 5700|    3|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      AU|2021-02-01|cntry_cd=AU/bus_d...|
|AU700716079381819...| 5700|    1|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      AU|2021-02-01|cntry_cd=AU/bus_d...|
+--------------------+-----+-----+-----+-----+-----+--------------------+-----------+--------------------+-----------+--------+----------+--------------------+


scala> transformedDF.write.format("org.apache.hudi").
     | options(getQuickstartWriteConfigs).
     | option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "col_9").
     | option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "col_2,col_1,col_3").
     | option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
     | option(DataSourceWriteOptions.KEYGENERATOR_CLASS_OPT_KEY, "org.apache.hudi.keygen.ComplexKeyGenerator").
     | option("hoodie.upsert.shuffle.parallelism","2").
     | option("hoodie.insert.shuffle.parallelism","2").
     | option(HoodieWriteConfig.TABLE_NAME, "targetTableHudi").
     | mode(SaveMode.Append).
     | save(targetPath)
                                                                                
scala> formattedDF.printSchema
root
 |-- col_1: string (nullable = true)
 |-- col_2: integer (nullable = true)
 |-- col_3: short (nullable = true)
 |-- col_4: string (nullable = true)
 |-- col_5: string (nullable = true)
 |-- col_6: byte (nullable = true)
 |-- col_9: timestamp (nullable = true)
 |-- col_10: string (nullable = true)
 |-- col_11: timestamp (nullable = true)
 |-- col_12: string (nullable = true)
 |-- cntry_cd: string (nullable = true)
 |-- bus_dt: date (nullable = true)


scala> formattedDF.show
+--------------------+-----+-----+-----+-----+-----+--------------------+-----------+--------------------+-----------+--------+----------+
|               col_1|col_2|col_3|col_4|col_5|col_6|               col_9|     col_10|              col_11|     col_12|cntry_cd|    bus_dt|
+--------------------+-----+-----+-----+-----+-----+--------------------+-----------+--------------------+-----------+--------+----------+
|7IN00716079317820...|  716|    3|   AB|  INR| null|2021-02-14 20:23:...|useridjsb91|2021-02-14 20:23:...|useridjsb91|      IN|2021-02-01|
|7IN00716079317820...|  716|    2|   AB|  INR| null|2021-02-14 20:23:...|useridjsb91|2021-02-14 20:23:...|useridjsb91|      IN|2021-02-01|
|7IN00716079317820...|  716|    1|   AB|  INR| null|2021-02-14 20:23:...|useridjsb91|2021-02-14 20:23:...|useridjsb91|      IN|2021-02-01|
|AU700716079381819...| 5700|    5|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      IN|2021-02-02|
|AU700716079381819...| 5700|    6|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      IN|2021-02-02|
|AU700716079381819...| 5700|    4|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      AU|2021-02-01|
|AU700716079381819...| 5700|    3|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      AU|2021-02-01|
|AU700716079381819...| 5700|    1|   AB|  INR| null|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91|      AU|2021-02-01|
+--------------------+-----+-----+-----+-----+-----+--------------------+-----------+--------------------+-----------+--------+----------+


scala> val testSnapshotDF = spark.read.format("org.apache.hudi").
     | load(targetPath + "/*/*/*")
21/02/24 14:36:27 WARN DefaultSource: Loading Base File Only View.
testSnapshotDF: org.apache.spark.sql.DataFrame = [_hoodie_commit_time: string, _hoodie_commit_seqno: string ... 16 more fields]

scala> testSnapshotDF.createOrReplaceTempView("test_hudi_ro")

scala> spark.sql("select cntry_cd,bus_dt,count(1) from test_hudi_ro group by cntry_cd,bus_dt").show
+--------+----------+--------+
|cntry_cd|    bus_dt|count(1)|
+--------+----------+--------+
|      IN|2021-02-01|       3|
|      AU|2021-02-01|       3|
|      IN|2021-02-02|       2|
+--------+----------+--------+

[bvaradar]

I was unable to setup spark-2.3.0 in my setup. But,with spark-2.4.4, this works fine as below. Can you use spark-2.4.x version. spark-2.3 seems too old though ?

21/02/25 05:14:26 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties Setting default log level to "WARN". To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel). Spark context Web UI available at http://ip-192-168-1-81.ec2.internal:4040 Spark context available as 'sc' (master = local[*], app id = local-1614258873363). Spark session available as 'spark'. Welcome to ____ __ / __/__ ___ _____/ /__ _\ \/ _ \/ _ / / '/
// .__/_,// //_\ version 2.4.4
//

Using Scala version 2.11.12 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_251)
Type in expressions to have them evaluated.
Type :help for more information.

scala> import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.DataSourceWriteOptions

scala> import org.apache.hudi.QuickstartUtils.getQuickstartWriteConfigs
import org.apache.hudi.QuickstartUtils.getQuickstartWriteConfigs

scala> import org.apache.hudi.config.{HoodieCompactionConfig, HoodieStorageConfig, HoodieWriteConfig}
import org.apache.hudi.config.{HoodieCompactionConfig, HoodieStorageConfig, HoodieWriteConfig}

scala> import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.{SaveMode, SparkSession}

scala> import org.apache.spark.sql.functions.{col, concat, lit}
import org.apache.spark.sql.functions.{col, concat, lit}

scala>

scala> val inputDF = spark.read.format("csv").option("header", "true").load("file:///tmp/input.csv")
inputDF: org.apache.spark.sql.DataFrame = [col_1: string, col_2: string ... 12 more fields]

scala> val formattedDF = inputDF.selectExpr("col_1", "cast(col_2 as integer) col_2",
| "cast(col_3 as short) col_3", "col_4", "col_5", "cast(col_6 as byte) col_6", "cast(col_7 as decimal(9,2)) col_7",
| "cast(col_8 as decimal(9,2)) col_8", "cast(col_9 as timestamp) col_9", "col_10", "cast(col_11 as timestamp) col_11",
| "col_12", "cntry_cd", "cast(bus_dt as date) bus_dt")
formattedDF: org.apache.spark.sql.DataFrame = [col_1: string, col_2: int ... 12 more fields]

scala> formattedDF.printSchema()
root
|-- col_1: string (nullable = true)
|-- col_2: integer (nullable = true)
|-- col_3: short (nullable = true)
|-- col_4: string (nullable = true)
|-- col_5: string (nullable = true)
|-- col_6: byte (nullable = true)
|-- col_7: decimal(9,2) (nullable = true)
|-- col_8: decimal(9,2) (nullable = true)
|-- col_9: timestamp (nullable = true)
|-- col_10: string (nullable = true)
|-- col_11: timestamp (nullable = true)
|-- col_12: string (nullable = true)
|-- cntry_cd: string (nullable = true)
|-- bus_dt: date (nullable = true)

scala> formattedDF.show
+--------------------+-----+-----+-----+-----+-----+-----+-----+--------------------+-----------+--------------------+-----------+--------+----------+
| col_1|col_2|col_3|col_4|col_5|col_6|col_7|col_8| col_9| col_10| col_11| col_12|cntry_cd| bus_dt|
+--------------------+-----+-----+-----+-----+-----+-----+-----+--------------------+-----------+--------------------+-----------+--------+----------+
|7IN00716079317820...| 716| 3| AB| INR| null| 0.00| 1.00|2021-02-14 20:23:...|useridjsb91|2021-02-14 20:23:...|useridjsb91| IN|2021-02-01|
|7IN00716079317820...| 716| 2| AB| INR| null| 0.00| 1.00|2021-02-14 20:23:...|useridjsb91|2021-02-14 20:23:...|useridjsb91| IN|2021-02-01|
|7IN00716079317820...| 716| 1| AB| INR| null| 0.00| 1.00|2021-02-14 20:23:...|useridjsb91|2021-02-14 20:23:...|useridjsb91| IN|2021-02-01|
|AU700716079381819...| 5700| 5| AB| INR| null| 0.00| 1.00|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91| IN|2021-02-02|
|AU700716079381819...| 5700| 6| AB| INR| null| 4.00| 1.97|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91| IN|2021-02-02|
|AU700716079381819...| 5700| 4| AB| INR| null| 0.00| 1.00|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91| AU|2021-02-01|
|AU700716079381819...| 5700| 3| AB| INR| null| 0.00| 1.00|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91| AU|2021-02-01|
|AU700716079381819...| 5700| 1| AB| INR| null| 0.00| 1.00|2021-02-14 20:06:...|useridjsb91|2021-02-14 20:06:...|useridjsb91| AU|2021-02-01|
+--------------------+-----+-----+-----+-----+-----+-----+-----+--------------------+-----------+--------------------+-----------+--------+----------+

scala>

scala> val transformedDF = formattedDF.withColumn("partitionpath", concat(lit("cntry_cd="), col("cntry_cd"), lit("/bus_dt="), col("bus_dt")))
transformedDF: org.apache.spark.sql.DataFrame = [col_1: string, col_2: int ... 13 more fields]

scala> val targetPath = "file:///tmp/output/"
targetPath: String = file:///tmp/output/

scala> transformedDF.write.format("org.apache.hudi").
| options(getQuickstartWriteConfigs).
| option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "col_9").
| option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "col_2,col_1,col_3").
| option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
| option(DataSourceWriteOptions.KEYGENERATOR_CLASS_OPT_KEY, "org.apache.hudi.keygen.ComplexKeyGenerator").
| option("hoodie.upsert.shuffle.parallelism","2").
| option("hoodie.insert.shuffle.parallelism","2").
| option(HoodieWriteConfig.TABLE_NAME, "targetTableHudi").
| mode(SaveMode.Append).
| save(targetPath)

scala>`

[codejoyan]

This is the spark version of the cluster being used at work so I will have to use Spark 2.3 until there is an upgrade. Since the documentation says: Hudi works with Spark-2.x, I was wondering there would be some workaround to this.

[bvaradar]

Sorry for the delay. I will try setting up 2.30 tonight to reproduce this issue.

[codejoyan]

Thanks @bvaradar

[bvaradar]

@codejoyan : Just realized this is because spark-2.3.0 uses com.databricks version of spark-avro. We moved to org.apache.spark:spark-avro.

Amazon EMR got this working by compiling hudi-spark-bundle with additional flag "-Dspark-shade-unbundle-avro". Additional context in https://issues.apache.org/jira/browse/HUDI-254.

@umehrot2 : Can you confirm if the above statement is correct ?

[umehrot2]

@codejoyan I think the problem stems because you are using org.apache.spark:spark-avro_2.11:2.4.4 in you packages with spark-submit. This is incompatible with spark-sql 2.3.0 that is why you get that NoSuchMethod exception.

What happens if you use databricks-avro package instead of spark-avro: https://mvnrepository.com/artifact/com.databricks/spark-avro ? Can you give that a shot ?

Also @bvaradar is right. When you build your Hudi and use databricks-avro package, please build you Hudi with -Pspark-shade-unbundle-avro to avoid bundling/shading of spark-avro module in hudi-spark-bundle.

[codejoyan]

Hi @umehrot2

I tried using databricks-avro along with avro 1.7.7 and avro 1.8.2 but I still get NoSuchMethod exceptions.
I am yet to try and build Hudi. Will check on that.

databricks-avro and avro 1.8.2

spark-shell \
> --packages org.apache.hudi:hudi-spark-bundle_2.11:0.7.0,com.databricks:spark-avro_2.11:4.0.0,org.apache.avro:avro:1.8.2 \
> --conf spark.driver.extraClassPath=/u/users/joyan/.ivy2/jars/com.databricks_spark-avro_2.11-4.0.0.jar,/u/users/j0s0j7j/.ivy2/jars/org.apache.avro_avro-1.8.2.jar \
> --conf spark.executor.extraClassPath=/u/users/joyan/.ivy2/jars/com.databricks_spark-avro_2.11-4.0.0.jar,/u/users/j0s0j7j/.ivy2/jars/org.apache.avro_avro-1.8.2.jar \
> --conf "spark.sql.hive.convertMetastoreParquet=false" \
> --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer'

java.lang.NoSuchMethodError: org.apache.avro.Schema.createUnion([Lorg/apache/avro/Schema;)Lorg/apache/avro/Schema;
  at org.apache.hudi.spark.org.apache.spark.sql.avro.SchemaConverters$.toAvroType(SchemaConverters.scala:185)
  at org.apache.hudi.spark.org.apache.spark.sql.avro.SchemaConverters$$anonfun$5.apply(SchemaConverters.scala:176)
  at org.apache.hudi.spark.org.apache.spark.sql.avro.SchemaConverters$$anonfun$5.apply(SchemaConverters.scala:174)
  at scala.collection.Iterator$class.foreach(Iterator.scala:893)
  at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
  at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
  at org.apache.spark.sql.types.StructType.foreach(StructType.scala:99)
  at org.apache.hudi.spark.org.apache.spark.sql.avro.SchemaConverters$.toAvroType(SchemaConverters.scala:174)
  at org.apache.hudi.AvroConversionUtils$.convertStructTypeToAvroSchema(AvroConversionUtils.scala:52)
  at org.apache.hudi.HoodieSparkSqlWriter$.write(HoodieSparkSqlWriter.scala:139)
  at org.apache.hudi.DefaultSource.createRelation(DefaultSource.scala:134)
  at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:46)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:70)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:68)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:86)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
  at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
  at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:80)
  at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:80)
  at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:654)
  at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:654)
  at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:77)
  at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:654)
  at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:273)
  at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:267)
  at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:225)
  ... 59 elided

databricks-avro and avro 1.7.7

spark-shell \
> --packages org.apache.hudi:hudi-spark-bundle_2.11:0.7.0,com.databricks:spark-avro_2.11:4.0.0,org.apache.avro:avro:1.7.7 \
> --conf spark.driver.extraClassPath=/u/users/joyan/.ivy2/jars/com.databricks_spark-avro_2.11-4.0.0.jar,/u/users/j0s0j7j/.ivy2/jars/org.apache.avro_avro-1.7.7.jar \
> --conf spark.executor.extraClassPath=/u/users/joyan/.ivy2/jars/com.databricks_spark-avro_2.11-4.0.0.jar,/u/users/j0s0j7j/.ivy2/jars/org.apache.avro_avro-1.7.7.jar \
> --conf "spark.sql.hive.convertMetastoreParquet=false" \
> --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer'

java.lang.NoClassDefFoundError: org/apache/avro/LogicalType
  at org.apache.hudi.AvroConversionUtils$.getAvroRecordNameAndNamespace(AvroConversionUtils.scala:71)
  at org.apache.hudi.HoodieSparkSqlWriter$.write(HoodieSparkSqlWriter.scala:135)
  at org.apache.hudi.DefaultSource.createRelation(DefaultSource.scala:134)
  at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:46)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:70)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:68)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:86)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
  at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
  at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:80)
  at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:80)
  at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:654)
  at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:654)
  at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:77)
  at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:654)
  at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:273)
  at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:267)
  at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:225)
  ... 59 elided
Caused by: java.lang.ClassNotFoundException: org.apache.avro.LogicalType
  at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
  at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
  at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
  ... 81 more

[codejoyan]

Unfortunately even after building the file and using -Pspark-shade-unbundle-avro I am running into the same issue. Am I missing anything

Build command:
mvn -DskipTests -Pspark-shade-unbundle-avro clean package

Spark command

spark-shell \
> --jars hudi-spark-bundle_2.11-0.7.0.jar,/u/users/j0s0j7j/.ivy2/jars/org.apache.spark_spark-avro_2.11-2.4.4.jar,/u/users/joyan/.ivy2/jars/org.apache.avro_avro-1.8.2.jar \
> --conf spark.driver.extraClassPath=/u/users/joyan/.ivy2/jars/org.apache.avro_avro-1.8.2.jar:/u/users/joyan/.ivy2/jars/org.apache.spark_spark-avro_2.11-2.4.4.jar \
> --conf spark.executor.extraClassPath=/u/users/joyan/.ivy2/jars/org.apache.avro_avro-1.8.2.jar:/u/users/joyan/.ivy2/jars/org.apache.spark_spark-avro_2.11-2.4.4.jar \
> --conf "spark.sql.hive.convertMetastoreParquet=false" \
> --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer'

Error Message

scala> transformedDF.write.format("org.apache.hudi").
     | options(getQuickstartWriteConfigs).
     | option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "col_9").
     | option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "col_2,col_1,col_3").
     | option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
     | option(DataSourceWriteOptions.KEYGENERATOR_CLASS_OPT_KEY, "org.apache.hudi.keygen.ComplexKeyGenerator").
     | option("hoodie.upsert.shuffle.parallelism","2").
     | option("hoodie.insert.shuffle.parallelism","2").
     | option(HoodieWriteConfig.TABLE_NAME, "targetTableHudi").
     | mode(SaveMode.Append).
     | save(targetPath)
java.lang.NoSuchMethodError: org.apache.spark.sql.types.Decimal$.minBytesForPrecision()[I
  at org.apache.spark.sql.avro.SchemaConverters$.toAvroType(SchemaConverters.scala:156)
  at org.apache.spark.sql.avro.SchemaConverters$$anonfun$5.apply(SchemaConverters.scala:176)
  at org.apache.spark.sql.avro.SchemaConverters$$anonfun$5.apply(SchemaConverters.scala:174)
  at scala.collection.Iterator$class.foreach(Iterator.scala:893)
  at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
  at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
  at org.apache.spark.sql.types.StructType.foreach(StructType.scala:99)
  at org.apache.spark.sql.avro.SchemaConverters$.toAvroType(SchemaConverters.scala:174)
  at org.apache.hudi.AvroConversionUtils$.convertStructTypeToAvroSchema(AvroConversionUtils.scala:52)
  at org.apache.hudi.HoodieSparkSqlWriter$.write(HoodieSparkSqlWriter.scala:139)
  at org.apache.hudi.DefaultSource.createRelation(DefaultSource.scala:134)
  at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:46)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:70)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:68)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:86)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
  at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
  at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:80)
  at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:80)
  at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:654)
  at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:654)
  at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:77)
  at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:654)
  at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:273)
  at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:267)
  at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:225)
  ... 59 elided

[nsivabalan]

@bvaradar @umehrot2 : reminder to follow up on this issue. Also, once resolved, can you please add a faq entry w/o fail. once you respond, can you please remove "awaiting-community-help" label for the issue and add "awaiting-user-response" label. Or close out the ticket if resolved and all AIs are done(faq entry etc)