用spark把mysql存入hudi

将 MySQL 中的数据存储到 Hudi 中，可以通过以下步骤实现：

1. 使用 Spark SQL 将 MySQL 中的数据读取到 Spark DataFrame 中。

2. 使用 Hudi 提供的 API 将 DataFrame 中的数据写入到 Hudi 表中。

下面是一个示例代码，可以将 MySQL 中的数据存储到 Hudi 中：

import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig._

val spark = SparkSession.builder()
  .appName("MySQL to Hudi")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .config("spark.sql.hive.convertMetastoreParquet", "false")
  .enableHiveSupport()
  .getOrCreate()

// 从 MySQL 中读取数据
val mysqlDF = spark.read.format("jdbc")
  .option("url", "jdbc:mysql://localhost:3306/test")
  .option("dbtable", "test_table")
  .option("user", "root")
  .option("password", "123456")
  .load()

// 将数据写入到 Hudi 中
mysqlDF.write.format("org.apache.hudi")
  .options(Map(
    TABLE_TYPE_OPT_KEY -> "MERGE_ON_READ",
    RECORDKEY_FIELD_OPT_KEY -> "id",
    PRECOMBINE_FIELD_OPT_KEY -> "timestamp",
    PARTITIONPATH_FIELD_OPT_KEY -> "date",
    KEYGENERATOR_CLASS_OPT_KEY -> "org.apache.hudi.keygen.NonpartitionedKeyGenerator",
    HIVE_SYNC_ENABLED_OPT_KEY -> "true",
    HIVE_DATABASE_OPT_KEY -> "default",
    HIVE_TABLE_OPT_KEY -> "hudi_table",
    HIVE_PARTITION_FIELDS_OPT_KEY -> "date",
    HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY -> "org.apache.hudi.hive.NonPartitionedExtractor",
    HIVE_STYLE_PARTITIONING_OPT_KEY -> "true",
    HIVE_ASSUME_DATE_PARTITION_OPT_KEY -> "false"
  ))
  .mode("append")
  .save("/path/to/hudi_table")

在上面的代码中，我们使用 Spark SQL 从 MySQL 中读取数据，并将其保存到 DataFrame 中。然后，我们使用 Hudi 提供的 API 将 DataFrame 中的数据写入到 Hudi 表中。

在写入数据时，我们需要指定 Hudi 表的类型（在这里是“MERGE_ON_READ”）、记录键、预合并字段、分区路径键等参数。我们还需要指定 Hive 元数据信息，这些信息将用于将 Hudi 表注册到 Hive 中。

最后，我们将数据保存到指定的 Hudi 表路径中。注意，这里使用的是“append”模式，这意味着我们将向现有的 Hudi 表中添加新数据。如果您想覆盖现有的数据，请使用“overwrite”模式。