代码拉取完成,页面将自动刷新
package com.dkl.blog.hudi
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.QuickstartUtils.getQuickstartWriteConfigs
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.spark.sql.SparkSession
/**
* Created by dongkelun on 2021/7/10 16:39
*
* 测试预合并
*/
object TestHuDiPreCombinedFiled {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder()
.master("local[*]")
.appName("TestHuDiPreCombinedFiled")
.config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
.getOrCreate()
testProgram(spark)
testSQLNoPreCombineFiled(spark)
testSQLWithPreCombineFiled(spark)
testSQLWithPreCombineFiled(spark)
spark.stop()
}
def testProgram(spark: SparkSession) {
val tableName = "test_hudi_table1"
val data = Array((7, "name12", 1.21, 108L, "2021-05-06"), (7, "name2", 2.22, 108L, "2021-05-06"),
(7, "name3", 3.45, 108L, "2021-05-06")
)
val df = spark.createDataFrame(data).toDF("id", "name", "price", "ts", "dt")
df.write.format("hudi").
options(getQuickstartWriteConfigs).
option(PRECOMBINE_FIELD.key(), "ts"). //指定preCombinedField=ts
option(RECORDKEY_FIELD.key(), "id").
option(PARTITIONPATH_FIELD.key(), "dt").
option(HIVE_STYLE_PARTITIONING.key(), true). //hive 分区路径的格式是否和hive一样,如果true,则:分区字段=
option("hoodie.table.name", tableName).
// option("hoodie.datasource.write.insert.drop.duplicates", true). //不更新
// option(OPERATION.key(), "INSERT").
option(HoodieWriteConfig.KEYGENERATOR_CLASS_NAME.key(), "org.apache.hudi.keygen.ComplexKeyGenerator").
mode("append").
save(s"/tmp/${tableName}")
val read_df = spark.
read.
format("hudi").
load(s"/tmp/${tableName}" + "/*")
read_df.show()
}
def testSQLNoPreCombineFiled(spark: SparkSession): Unit = {
val tableName = "test_hudi_table2"
spark.sql(
s"""
| create table ${tableName} (
| id int,
| name string,
| price double,
| ts long,
| dt string
|) using hudi
| partitioned by (dt)
| options (
| primaryKey = 'id',
| type = 'cow'
| )
| location '/tmp/${tableName}'
|""".stripMargin)
spark.sql(s"show create table ${tableName}").show(false)
spark.sql(
s"""
|merge into ${tableName} as t0
|using (
| select 1 as id, 'hudi' as name, 97 as price, 99 as ts, '2021-05-05' as dt,'INSERT' as opt_type union
| select 1 as id, 'hudi_2' as name, 98 as price, 99 as ts, '2021-05-05' as dt,'UPDATE' as opt_type union
| select 1 as id, 'hudi_2' as name, 99 as price, 99 as ts, '2021-05-05' as dt,'UPDATE' as opt_type union
| select 3 as id, 'hudi' as name, 10 as price, 110 as ts, '2021-05-05' as dt ,'DELETE' as opt_type
| ) as s0
|on t0.id = s0.id
|when matched and opt_type!='DELETE' then update set *
|when matched and opt_type='DELETE' then delete
|when not matched and opt_type!='DELETE' then insert *
|""".stripMargin)
spark.table(tableName).show()
}
def testSQLWithPreCombineFiled(spark: SparkSession): Unit = {
val tableName = "test_hudi_table3"
spark.sql(
s"""
| create table ${tableName} (
| id int,
| name string,
| price double,
| ts long,
| dt string
|) using hudi
| partitioned by (dt)
| options (
| primaryKey = 'id',
| preCombineField = 'ts',
| type = 'cow'
| )
| location '/tmp/${tableName}'
|""".stripMargin)
spark.sql(s"show create table ${tableName}").show(false)
spark.sql(
s"""
|merge into ${tableName} as t0
|using (
| select 1 as id, 'hudi' as name, 97 as price, 99 as ts, '2021-05-05' as dt,'INSERT' as opt_type union
| select 1 as id, 'hudi_2' as name, 98 as price, 99 as ts, '2021-05-05' as dt,'UPDATE' as opt_type union
| select 1 as id, 'hudi_2' as name, 99 as price, 99 as ts, '2021-05-05' as dt,'UPDATE' as opt_type union
| select 3 as id, 'hudi' as name, 10 as price, 110 as ts, '2021-05-05' as dt ,'DELETE' as opt_type
| ) as s0
|on t0.id = s0.id
|when matched and opt_type!='DELETE' then update set *
|when matched and opt_type='DELETE' then delete
|when not matched and opt_type!='DELETE' then insert *
|""".stripMargin)
spark.table(tableName).show()
}
def testSQLAndProgram(spark: SparkSession): Unit = {
val tableName = "test_hudi_table4"
spark.sql(
s"""
| create table ${tableName} (
| id int,
| name string,
| price double,
| dt string
|) using hudi
| partitioned by (dt)
| options (
| primaryKey = 'id',
| type = 'cow'
| )
| location '/tmp/${tableName}'
|""".stripMargin)
val data = Array((7, "name12", 1.21, 106L, "2021-05-06"), (7, "name2", 2.22, 108L, "2021-05-06"),
(7, "name3", 3.45, 107L, "2021-05-06")
)
val df = spark.createDataFrame(data).toDF("id", "name", "price", "ts", "dt")
df.show()
df.write.format("hudi").
options(getQuickstartWriteConfigs).
option(PRECOMBINE_FIELD.key(), "ts").
option(RECORDKEY_FIELD.key(), "id").
option(PARTITIONPATH_FIELD.key(), "dt").
option(HIVE_STYLE_PARTITIONING.key(), true). //hive 分区路径的格式是否和hive一样,如果true,则:分区字段=
option("hoodie.table.name", tableName).
// option("hoodie.datasource.write.insert.drop.duplicates", true). //不更新
// option(OPERATION.key(), "INSERT").
option(HoodieWriteConfig.KEYGENERATOR_CLASS_NAME.key(), "org.apache.hudi.keygen.ComplexKeyGenerator").
save(s"/tmp/${tableName}")
val read_df = spark.
read.
format("hudi").
load(s"/tmp/${tableName}" + "/*")
read_df.show()
spark.table(tableName).show()
}
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。