from findspark import init init() import time import pymysql from pyspark import SparkContext from pyspark.streaming import StreamingContext sc = SparkContext(appName="mysql_streaming", master="local[*]") ssc = StreamingContext(sc, 10) # 设置检查点 ssc.checkpoint("/usr/local/work") # 累加器(状态更新) def updateFunction(newValues, runningCount): if runningCount is None: runningCount = 0 return sum(newValues, runningCount) # 设置套接字流 ############### Begin ############### ############### End ############### pairs = inputStream.flatMap(lambda x: x.split(" ")).filter(lambda x: x != "").map(lambda word: (word, 1)) wordCounts = pairs.updateStateByKey(updateFunction) wordCounts.pprint(100) def dbfunc(records): # 根据传入的 records 参数，完成数据写入 Mysql 操作 ############### Begin ############### ############### End ############### # 分区设置 def func(rdd): repartitionedRDD = rdd.repartition(3) repartitionedRDD.foreachPartition(dbfunc) wordCounts.foreachRDD(func=func) ssc.start() time.sleep(60) ssc.stop()

这段代码的作用是从一个套接字流中读取数据，进行词频统计，并将统计结果写入到 MySQL 数据库中。

以下是补充的代码：

# 设置套接字流
inputStream = ssc.socketTextStream("localhost", 8888)

# 数据库连接信息
db_config = {
    "host": "localhost",
    "port": 3306,
    "user": "root",
    "password": "password",
    "db": "test",
    "charset": "utf8mb4"
}

# 数据库连接
conn = pymysql.connect(**db_config)

def dbfunc(records):
    # 根据传入的 records 参数，完成数据写入 Mysql 操作
    cursor = conn.cursor()
    for record in records:
        word, count = record
        try:
            cursor.execute("INSERT INTO word_count(word, count) VALUES (%s, %s)", (word, count))
            conn.commit()
        except Exception as e:
            print(e)
            conn.rollback()
    cursor.close()

# ... 以下内容与原代码相同 ...

请注意，以上代码需要在 Spark 环境中执行，而不是在普通的 Python 环境中执行。你需要在服务器上安装 Spark 并以集群模式启动，然后在 Spark 集群中提交这段代码。

另外，需要在数据库中创建一个名为 `word_count` 的表，用于存储词频统计结果。表的结构如下：

CREATE TABLE word_count (
    id INT PRIMARY KEY AUTO_INCREMENT,
    word VARCHAR(255),
    count INT
);