既然前几期内容测试了 Flink 关于内存跟(MySQL)数据量之间的关系，那么这一次，在同等条件下，我们也来看看如果使用 Spark，那么它需要的内存，跟导出(MySQL)数据量之间，又一种什么样的关系呢？

之前的内容就说过，对于数据导入/导出这种事情，我一向秉承的原则就是，「能用 Flink 或者 Spark 来搞定的，就绝不用其他工具」。

原因很简单，这两者是组成大数据技术的最核心要素(至少我是这么认为的)，相对其他组件，是社区支持力度最大，迭代升级最频繁，生态最广、也是成熟度最高的，几乎没有之一。

这次测试的目的，在于生产上，如果面临类似的数据导入场景，可以给你提供一个选择的依据。

0. 准备

Spark 读取 MySQL，目前官方推荐的主流方法，有仅只有一个，那就是通过 jdbc 的方式，而相比于 Flink，它只能是「批处理」。

可能你会问：咦？Flink 读 MySQL 是流的方式(需要binlog支持)，而 Spark 读 MySQL ，就只能是离线方式，有什么可比性呢？

乍一看好像是，但你别忘了，在生产上，我们选择一个解决方案的关键，并不只是以「理论为驱动」，而更多会考虑解决方案的方便、高效、节省资源与否。

因为咱们这次的目的，是导出 MySQL 的「存量数据」，所以这里考虑用 Spark 以离线方式导出，也就有了合理性。

引入必要依赖

Spark 的 jdbc 方案，一般情况下，需要的依赖，相比 Flink 要更简单一点，这里要读 MySQL，就只需要引入 MySQL 自己的 jdbc 包就足够。

不存在跟 Flink 一样，还需要引入 Spark jdbc 的包一说。

然后，就可以编码了，简单的数据处理流程如下：

1. 编码

相比之前测试的 Flink CDC 3.x 全程只需要通过配置文件，就能实现对 MySQL 数据导入到 Kafka 来说，当前 Spark 需要编码的这种方式，确实要更麻烦一点。

但当你写完后会发现，其实还好，因为代码都很简单。

package com.anryg.bigdata.streaming.mysql

import com.alibaba.fastjson.JSONObject
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

/**
  * @DESC: Spark 读取MySQL写Kafka，对比跟Flink消耗的内存情况
  * @Auther: Anryg
  * @Date: 2024/8/27 15:09
  */
object SparkReadMysql2Kafka {

    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("SparkReadMysql2Kafka").setMaster("local[*]")
        val spark = SparkSession.builder().config(conf).getOrCreate()

        /**读取MySQL数据*/
        val rawDF = spark.read
            .format("jdbc")
            .option("url", "jdbc:mysql://192.168.221.132:3306/test")
            .option("driver", "com.mysql.cj.jdbc.Driver")
            .option("user", "pcl")
            .option("password", "****")
            .option("dbtable", "test01")
            .load()

        import spark.implicits._  /**引入隐式转换对象*/

        /**因为要写Kafka，需要对数据做一点必要的转换*/
        val targetDF = rawDF.map(row => {
                        val jsonObj = new JSONObject(26)
                        val subject_common_name = row.getString(0)
                        jsonObj.put("subject_common_name",subject_common_name)
                        val serial_number = row.getString(1)
                        jsonObj.put("serial_number", serial_number)
                        /**中间字段省略*/
                        val created_at = row.getTimestamp(25)
                        jsonObj.put("created_at",created_at)
                        jsonObj.toJSONString
                    }).toDF("value") /**写到Kafka，key跟value字段，至少需要提供value*/

        /**将转换后，符合Kafka要求的格式数据写进Kafka*/
        targetDF
            .write
            .format("kafka")
            .option("kafka.bootstrap.servers", "192.168.211.107:6667")
            .option("topic", "test")
            .save()
    }
}