spark streaming 用到mysql数据怎么加载？

可以使用JDBC（Java数据库连接）来将MySQL数据源加载到Spark Streaming中。具体来说，可以通过以下步骤实现：

1. 在Spark Streaming中创建一个StreamingContext对象。

2. 创建一个JDBC连接，连接到MySQL数据库。

3. 使用Spark Streaming内置的DStream对象创建一个输入流，该输入流将使用提供的数据源从MySQL数据库中读取数据。

4. 对每个批次的数据执行必要的转换和操作。

5. 将数据写回MySQL数据库。

下面是一个示例代码：

from pyspark.streaming import StreamingContext
from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext
import datetime

conf = SparkConf().setAppName("MySQLStream")
sc = SparkContext(conf=conf)
sqlContext = SQLContext(sc)
ssc = StreamingContext(sc, batchDuration=1)

# 创建JDBC连接
jdbcUrl = "jdbc:mysql://localhost:3306/mydatabase"
jdbcUsername = "username"
jdbcPassword = "password"

# 使用DStream对象创建输入流
def createMySQLStream():
    inputDF = sqlContext.read.format("jdbc").options(url=jdbcUrl, dbtable="mytable", user=jdbcUsername, password=jdbcPassword).load()
    stream = inputDF.writeStream.outputMode("append").format("console").start()
    return stream

# 对每个批次的数据执行必要的转换和操作
def processStream(batchTime, rdd):
    if not rdd.isEmpty():
        # 在这里处理批次的数据
        processedData = rdd
        # 将数据写回MySQL数据库
        processedData.write.jdbc(url=jdbcUrl, table="outputtable", mode="append", properties={"user": jdbcUsername, "password": jdbcPassword})

# 创建DStream对象，并开始流处理
stream = createMySQLStream()
stream.foreachRDD(processStream)
ssc.start()
ssc.awaitTermination()

注意：这只是一个简单的示例，实际生产环境中需要进行更多的错误处理和优化。