使用Spark抽取MySQL指定数据表中的增量数据到ods层的指定的分区表中

1. 从MySQL中读取增量数据

使用Spark SQL中的JDBC数据源，连接MySQL数据库，使用SQL语句读取增量数据。

val jdbcDF = spark.read
    .format("jdbc")
    .option("url", "jdbc:mysql://localhost/test")
    .option("dbtable", "table1")
    .option("user", "root")
    .option("password", "password")
    .option("driver", "com.mysql.jdbc.Driver")
    .load()

2. 将数据写入到指定分区表中

使用DataFrame API，将数据写入到ODS层的指定分区表中。

jdbcDF.write
    .partitionBy("date")
    .mode("append")
    .format("parquet")
    .save("hdfs://path/to/ods_table")

其中，date为分区字段，格式为yyyy-MM-dd。

完整代码示例：

import org.apache.spark.sql.SparkSession

object IncrementalDataTransfer {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("IncrementalDataTransfer")
      .master("local")
      .getOrCreate()

    // 从MySQL中读取增量数据
    val jdbcDF = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost/test")
      .option("dbtable", "table1")
      .option("user", "root")
      .option("password", "password")
      .option("driver", "com.mysql.jdbc.Driver")
      .load()

    // 将数据写入到指定分区表中
    jdbcDF.write
      .partitionBy("date")
      .mode("append")
      .format("parquet")
      .save("hdfs://path/to/ods_table")

    spark.stop()
  }
}