我会这样做：步骤1：对DataFrame进行分区

num_partitions = 10 
partitions = dataframe.randomSplit([1.0] * num_partitions)

第2步：为SQL Server和Oracle定义插入函数

from pyspark.sql import DataFrame

def insert_to_sql_server(df: DataFrame):
    jdbc_url = "jdbc:sqlserver://YOUR_SQL_SERVER_HOST:1433;databaseName=YOUR_DB_NAME;"
    properties = {
        "user": "YOUR_USERNAME",
        "password": "YOUR_PASSWORD",
        "driver": "com.microsoft.sqlserver.jdbc.SQLServerDriver"
    }
    df.write.jdbc(jdbc_url, "YOUR_TABLE_NAME", mode="append", properties=properties)

def insert_to_oracle(df: DataFrame):
    jdbc_url = "jdbc:oracle:thin:@YOUR_ORACLE_HOST:1521:YOUR_SID"
    properties = {
        "user": "YOUR_USERNAME",
        "password": "YOUR_PASSWORD",
        "driver": "oracle.jdbc.driver.OracleDriver"
    }
    df.write.jdbc(jdbc_url, "YOUR_TABLE_NAME", mode="append", properties=properties)

步骤3：使用ThreadPool并行插入

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=num_partitions) as executor:
    futures = [executor.submit(insert_to_sql_server, partition) for partition in partitions]
    for future in futures:
        future.result()

with ThreadPoolExecutor(max_workers=num_partitions) as executor:
    futures = [executor.submit(insert_to_oracle, partition) for partition in partitions]
    # Ensure all futures are done
    for future in futures:
        future.result()