如何降低Python编写的Spark应用CPU占用率？

应用架构 #

• ws_server 推送JSON数据
• Spark Socket Streaming 接收JSON数据
• Spark应用调用API获取数据并回写业务系统

现状与问题 #

应用功能正常，Spark能实时监听并传输正确数据，但CPU占用率持续100%。未发现单步操作存在高CPU消耗，怀疑是配置不合理、隐式循环或资源竞争导致。

运行方式 #

最初通过终端单独启动：

python3 ws_server.py
python3 spark_app.py

后尝试通过spark-submit加载配置文件运行：

spark-submit spark_app.py --conf "spark_defaults.conf"

注：配置文件已验证可正常读取。

相关代码与配置 #

ws_server.py #

import socket
import sys
import time
import json
import argparse
import datetime


def get_message():
    with open('path_to_json/data.json',
              'r') as file:
        json_str = file.read()
    dict_data = json.loads(json_str)
    data_list = list(dict_data.values())
    return " ".join(data_list) + "\n"

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Streaming generation socket server")
    parser.add_argument("port", type=int, help="port")
    args = parser.parse_args()

    port = args.port

    print(port)

    HOST = "127.0.0.1"
    PORT = port

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    print('Socket created')

    try:
        sock.bind((HOST, PORT))
    except socket.error as msg:
        print('Bind failed. Error Code : ' + str(msg))
        sys.exit()

    print('Socket bind complete')

    sock.listen(1)
    print('Listening on {}'.format(PORT))

    while True:
        try:
            conn, addr = sock.accept()
            print('Connected with ' + addr[0] + ':' + str(addr[1]))
        except KeyboardInterrupt:
            print("bye")
            sock.close()
            sys.exit()
        except:
            print("Error when accept connection")
            sock.close()
            sys.exit()

        while True:
            try:
                conn.send(get_message().encode())
                time.sleep(30)
            except KeyboardInterrupt:
                print("bye")
                sock.close()
                sys.exit()
            except:
                print("Client disconnect")
                break
    sock.close()
print("Done")

data.json #

{"value": "xxx_data,yyy_data"}

spark-defaults.conf #

spark.driver.memory              449m
spark.driver.cores               1
spark.executor.cores             1
spark.executor.memory            1g
spark.executor.instances         1
spark.dynamicAllocation.executorIdleTimeout      true
spark.storage.safetyFraction 0.9
spark.storage.memoryFraction 0.6

spark_app.py #

from pyspark.sql import SparkSession
from pyspark import StorageLevel
from pyspark.sql.functions import explode, split, udf, col, to_timestamp
import requests, json
from sqlalchemy import create_engine



def add_data_to_service(#it's working, private info):
    url = '200 response all good.com/api/v2/get_needed_data'
    headers = {'Content-Type': 'application/json'}
    params = {
        'its also': 'good'
    }
    res = requests.patch(url, json.dumps(params), headers=headers)
    print('status code', res.status_code) #always 200 and putting data successfully


def split_data(df):
    # Split column into two columns
    split_cols = split(df['data_together with , separator'], ',')
    new_df = df.withColumn('some data1', split_cols.getItem(0)) \
        .withColumn('some data2', split_cols.getItem(1)) \
        .withColumn('timestamp', split_cols.getItem(2))
    return new_df


def execute_Request(number: str):
    url = 'it's also working fine'
    r = requests.get(url,
                     params={
                         'needed params': 'needed params',
                         'api_key': 'its private'
                     },
                     headers={
                         'Content-Type': 'application/json'
                     })
    response = r.json()
    print(response)
    if r != None and r.status_code == 200:
        return response['what i need']
    else:
        return None


def split_response(df):
    split_cols = split(df['response'], ',')
    new_df = df.withColumn('data1', split_cols.getItem(0)) \
        .withColumn('data2', split_cols.getItem(1)) \
        .withColumn('data3', split_cols.getItem(2)) \
        .withColumn('data4', split_cols.getItem(3))
    return new_df


def transform(lines):
    df_streamed = lines.select(
        explode(
            split(lines.value, " ")
        ).alias("data_together with , separator")
    )
    # times data is streamed
    wordCounts = df_streamed.groupBy("data_together with , separator").count()
    # splitting data from socket
    splitted_data = split_data(wordCounts)
    no_duplicates_df = splitted_data.select("*").where("count < 2")
    # prepare execution
    executeRestApiUDF = udf(execute_Request)
    # collecting response data
    df_with_request_data = no_duplicates_df.withColumn("response", executeRestApiUDF(col("contact")))
    final_df = split_response(df_with_request_data)
    return final_df


def load_data(df, epoch_id):
    df_pandas = df.toPandas()
    if (df_pandas.empty):
        pass
    else:
        lead = str(df_pandas['data1'].iloc[-1])
        city = str(df_pandas['data2'].iloc[-1])
        country = str(df_pandas['data3'].iloc[-1])
        region = str(df_pandas['data4'].iloc[-1])
        timezone = int(df_pandas['data5'].iloc[-1])
        add_data_to_service(data1, data2, data3, data4, data5)
        # df_dropped = df.drop(columns=['B', 'C'])


def main(spark: SparkSession):
    spark.sparkContext.setLogLevel("ERROR")

    print('we start\n')
    lines = spark \
        .readStream \
        .format("socket") \
        .option("host", "127.0.0.1") \
        .option("port", 8069) \
        .load()

    transformed_df = transform(lines)
    query = transformed_df.writeStream.outputMode("update").foreachBatch(load_data).trigger(
        processingTime='10 seconds').start()
    query.awaitTermination()


if __name__ == '__main__':
    main(SparkSession
         .builder
         .config("spark.driver.memory", "449m")
         .config("spark.driver.cores", "1")
         .master("local[1]")
         .appName("test")
         .getOrCreate())

排查与优化方案 #

1. 解决Socket Source无等待轮询问题 #

Spark Socket Source默认会持续轮询Socket连接，即使无新数据也会不断尝试读取，这是CPU占用100%的核心原因之一。添加readTimeout参数限制轮询频率：

lines = spark \
    .readStream \
    .format("socket") \
    .option("host", "127.0.0.1") \
    .option("port", 8069) \
    .option("readTimeout", 10000)  # 设置10秒超时，减少空轮询次数
    .load()

2. 优化UDF的同步阻塞问题 #

execute_Request是同步HTTP请求UDF，每个请求都会阻塞线程，且Python UDF在Executor中为单线程运行，易导致CPU空转。优化方向：

• 改用异步HTTP客户端（如aiohttp）改写UDF，减少阻塞时间
• 将普通UDF改为Pandas UDF，利用批量处理提升效率
• 把API请求逻辑移到foreachBatch中，用批量请求替代单条请求

3. 匹配触发频率与资源配置 #

当前设置processingTime='10 seconds'，但Driver和Executor仅分配1核，若批处理无法在10秒内完成，会导致任务堆积、CPU持续高负载。调整建议：

• 将触发间隔延长至30秒，与ws_server的推送频率对齐：processingTime='30 seconds'
• 根据机器资源增加核心数，比如将spark.driver.cores和spark.executor.cores改为2
• 关闭动态分配（spark.dynamicAllocation.enabled=false），单Executor场景下动态分配会增加额外调度开销

4. 移除不必要的聚合操作 #

transform函数中的groupBy.count()是为了去重，但Socket Source每次仅推送一条数据，该聚合完全多余，会增加无意义的计算开销。直接移除后简化逻辑：

def transform(lines):
    df_streamed = lines.select(
        explode(
            split(lines.value, " ")
        ).alias("data_together with , separator")
    )
    # 直接拆分数据，无需聚合去重
    splitted_data = split_data(df_streamed)
    executeRestApiUDF = udf(execute_Request)
    df_with_request_data = splitted_data.withColumn("response", executeRestApiUDF(col("contact")))
    final_df = split_response(df_with_request_data)
    return final_df

5. 优化foreachBatch中的数据转换 #

load_data中将Spark DataFrame转为Pandas DataFrame再取最后一行，会把整批数据拉到Driver端，增加Driver压力。改用Spark API直接获取数据：

def load_data(df, epoch_id):
    if df.count() == 0:
        pass
    else:
        # 用Spark API获取最后一行数据
        last_row = df.orderBy(col("timestamp").desc()).limit(1).collect()[0]
        lead = last_row['data1']
        city = last_row['data2']
        country = last_row['data3']
        region = last_row['data4']
        timezone = last_row['data5']
        add_data_to_service(lead, city, country, region, timezone)

内容的提问来源于stack exchange，提问作者Дмитрий Григорчук