# 创建实时写入Hudi的Spark Streaming应用
import org.apache.spark.sql.SparkSession
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig._

object FgeduHudiStreaming {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder()
.appName(“FgeduHudiStreaming”)
.config(“spark.serializer”, “org.apache.spark.serializer.KryoSerializer”)
.getOrCreate()

import spark.implicits._

val tableName = “fgedu_streaming_table”
val basePath = “hdfs://fgedu.net.cn:8020/bigdata/fgdata/hudi/fgedu_streaming_table”

// 模拟读取Kafka数据
val kafkaDF = spark.readStream
.format(“kafka”)
.option(“kafka.bootstrap.servers”, “192.168.1.100:9092,192.168.1.101:9092”)
.option(“subscribe”, “fgedu_events”)
.load()

// 解析JSON数据
val parsedDF = kafkaDF.selectExpr(“CAST(value AS STRING)”)
.select(from_json($”value”, schema).as(“data”))
.select(“data.*”)

// 写入Hudi
val query = parsedDF.writeStream
.format(“hudi”)
.option(TABLE_NAME, tableName)
.option(RECORDKEY_FIELD_OPT_KEY, “id”)
.option(PRECOMBINE_FIELD_OPT_KEY, “ts”)
.option(PARTITIONPATH_FIELD_OPT_KEY, “dt”)
.option(“hoodie.datasource.write.operation”, “upsert”)
.option(“checkpointLocation”, “/bigdata/fgdata/checkpoints/fgedu_streaming”)
.start(basePath)

query.awaitTermination()
}
}

# 提交应用
$ spark-submit –class FgeduHudiStreaming \
–packages org.apache.hudi:hudi-spark3.3-bundle_2.12:0.14.0 \
–master yarn \
–deploy-mode cluster \
–num-executors 10 \
–executor-memory 16G \
–executor-cores 4 \
fgedu-hudi-streaming.jar

# 配置MoR表
val tableName = “fgedu_mor_table”
val basePath = “hdfs://fgedu.net.cn:8020/bigdata/fgdata/hudi/fgedu_mor_table”

# 写入MoR表
df.write.format(“hudi”).
option(TABLE_NAME, tableName).
option(RECORDKEY_FIELD_OPT_KEY, “uuid”).
option(PRECOMBINE_FIELD_OPT_KEY, “ts”).
option(PARTITIONPATH_FIELD_OPT_KEY, “partitionpath”).
option(WRITE_OPERATION_OPT_KEY, “insert”).
option(TABLE_TYPE_OPT_KEY, “MERGE_ON_READ”).
option(COMPACTION_INLINE_OPT_KEY, “true”).
option(COMPACTION_INLINE_MAX_DELTA_COMMITS_OPT_KEY, “5”).
mode(Overwrite).
save(basePath)

# 读取优化视图（只读Parquet）
val roViewDF = spark.read.format(“hudi”).
option(QUERY_TYPE_OPT_KEY, QUERY_TYPE_READ_OPTIMIZED_OPT_VAL).
load(basePath)
roViewDF.show()

# 读取实时视图（合并日志文件）
val rtViewDF = spark.read.format(“hudi”).
option(QUERY_TYPE_OPT_KEY, QUERY_TYPE_SNAPSHOT_OPT_VAL).
load(basePath)
rtViewDF.show()

# 创建Hive外部表
CREATE EXTERNAL TABLE fgedu_hudi_hive_table (
uuid string,
partitionpath string,
rider string,
begin_lon double,
begin_lat double,
ts bigint
)
STORED AS INPUTFORMAT ‘org.apache.hudi.hadoop.hive.HoodieParquetInputFormat’
OUTPUTFORMAT ‘org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat’
LOCATION ‘hdfs://fgedu.net.cn:8020/bigdata/fgdata/hudi/fgedu_hudi_table’;

# 查询Hudi表
SELECT * FROM fgedu_hudi_hive_table LIMIT 10;

# 使用Presto查询Hudi表
# 在Presto中配置Hudi连接器
SELECT * FROM hudi.fgedu_hudi_table WHERE partitionpath = ‘2024/01/01’ LIMIT 10;

# 查询优化建议
# 1. 使用分区过滤
# 2. 使用RecordKey过滤
# 3. 使用增量查询
# 4. 合理设置并行度
# 5. 定期执行Compaction和Clustering