Hudi 支持 `update` 操作？

Question

使用 hudi 的 spark sql 更新记录时出现异常，如下所示。

update hudi.cow1 set price=1300 where id=2;
22/10/17 19:24:44 ERROR Executor: Exception in task 0.0 in stage 206.0 (TID 2442)
org.apache.avro.AvroRuntimeException: Not a valid schema field:
    at org.apache.avro.generic.GenericData$Record.get(GenericData.java:256)
    at org.apache.hudi.avro.HoodieAvroUtils.getNestedFieldVal(HoodieAvroUtils.java:503)
    at org.apache.hudi.HoodieSparkSqlWriter$.$anonfun$write$11(HoodieSparkSqlWriter.scala:295)
    at scala.collection.Iterator$$anon$10.next(Iterator.scala:461)
    at scala.collection.Iterator$$anon$10.next(Iterator.scala:461)
    at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:199)
    at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:63)
    at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)
    at org.apache.spark.scheduler.Task.run(Task.scala:131)
    at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1491)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
22/10/17 19:24:44 WARN TaskSetManager: Lost task 0.0 in stage 206.0 (TID 2442) (192.168.2.228 executor driver): org.apache.avro.AvroRuntimeException: Not a valid schema field:
    at org.apache.avro.generic.GenericData$Record.get(GenericData.java:256)
    at org.apache.hudi.avro.HoodieAvroUtils.getNestedFieldVal(HoodieAvroUtils.java:503)
    at org.apache.hudi.HoodieSparkSqlWriter$.$anonfun$write$11(HoodieSparkSqlWriter.scala:295)
    at scala.collection.Iterator$$anon$10.next(Iterator.scala:461)
    at scala.collection.Iterator$$anon$10.next(Iterator.scala:461)
    at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:199)
    at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:63)
    at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)
    at org.apache.spark.scheduler.Task.run(Task.scala:131)
    at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1491)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

我创建了如下表。

  create table if not exists cow1 (
                      id int,
                      name string,
                      price double
                    ) using hudi
                    options (
                      type = 'cow',
                      primaryKey = 'id'
                    );

我的环境是： mac系统； 火花：spark-3.2.2-bin-hadoop3.2 hudi：hudi-spark3.2-bundle_2.12-0.12.0.jar 我把 hudi jar 放在 spark home 下的 jars dir 中。

我开始使用 spark sql：

./spark-sql --jars ../../hudi-spark3.2-bundle_2.12-0.12.0.jar \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
--conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
--conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog'

这是一个hudi错误吗？

Answer 1

我在github上问过。 update 需要 preComineKey 属性。