1. 通过spark-sql创建hudi表
   

create table if not exists hudi_table3(
  id int,
  name string,
  price double
)  using hudi
options (
  'type' = 'mor',
  'primaryKey' = 'id',
  'hoodie.datasource.hive_sync.enable'='false',
  'hoodie.datasource.meta.sync.enable'='false',
  'hoodie.datasource.write.precombine.field'=price
)

  2. 写入hudi代码

val spark =
  SparkSession.builder()
    .master("local[*]")
    .enableHiveSupport()
    .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .getOrCreate()
val df = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "test")
  .option("group.id","test-1")
  .load()
import spark.implicits._
val query = df
  .selectExpr("split(cast(value as string),',') as sp")
  .selectExpr("cast(sp[0] as int) as id ","sp[1] as name","cast(sp[2] as double) as price")
  .writeStream.format("hudi")
  .trigger(Trigger.ProcessingTime(5000L))
  .option("checkpointLocation","file:///Users/haoguangshi/mysoft/ck")
  .option("path","/Users/haoguangshi/workspace/hudi-lrn/spark-warehouse/hudi_table3")
  // 主键相同的话根据该字段进行判断需要保留那行数据PRECOMBINE_FIELD_OPT_KEY
  .option("hoodie.datasource.write.precombine.field","price")
  // 表主键 RECORDKEY_FIELD_OPT_KEY
  .option("hoodie.datasource.write.recordkey.field","id")
  .start()
query.awaitTermination()
spark.stop()