听起来瓶颈是从rdd->dataframes的转换，而且手头的函数相当快，通过pyarrow将pandadf转换为spark df也相当快。以下是两种可能的解决方案：
因为并行创建df很容易，所以使用 df.to_parquet ，即：

def generate_data(seed):
    M = 10
    np.random.seed(seed)
    np_array = np.random.random_sample(M) # generates an array of M random values
    df = pd.DataFrame(np_array, columns=["x"])
    df["seed"] = seed
    df.reset_index().to_parquet(f"s3://bucket/part-{str(seed).zfill(5)}.parquet"

Spark读取结果Parquet文件应该是微不足道的事后。然后瓶颈就变成了io限制，这应该比spark转换元组/行类型快。
如果不允许您将任何内容保存到文件中， pandas_udf 以及 GROUPED_MAP 如果你的spark版本足够新的话，可能会对你有所帮助。它还使用pyarrow在spark dfs和pandas dfs之间进行转换，因此它应该比使用元组更快，并允许您以分布式方式从udf创建和返回pandas dfs。

import numpy as np
import pandas as pd
from pyspark.sql.functions import pandas_udf, PandasUDFType

N = 10

df = spark.createDataFrame(
    [(i,) for i in range(N)], ["seed"]
)

def generate_data(seed):
    M = 10
    np.random.seed(seed)
    np_array = np.random.random_sample(M) # generates an array of M random values
    df = pd.DataFrame(np_array, columns=["x"])
    df["seed"] = seed
    return df.reset_index()

@pandas_udf("index long, x double, seed long", PandasUDFType.GROUPED_MAP)
def generate_data_udf(pdf):
    output = []
    for idx, row in pdf.iterrows():
        output.append(generate_data(row["seed"]))
    return pd.concat(output)

df.groupby("seed").apply(generate_data_udf).show()

较慢的部分将是 groupby 你可能可以加快速度，这取决于你如何批量种子进入 generate_data_udf ，即：

@udf(returnType=IntegerType())
def batch_seed(seed):
    return seed // 10

df.withColumn("batch_seed", batch_seed(col("seed"))). \
groupBy("batch_seed").apply(generate_data_udf).show()

以下是一个不使用rdd或创建行的解决方案，但仅使用Dataframe操作：
（代码是用scala编写的，但用python编写同样的代码应该很简单）

val N = 100000

//for seed return array of index and random_value
def generate_data(i: Int): Array[(Int, Double)] = ???
val generate_data_udf = udf (generate_data _)

spark
  .range(N)
  .toDF("seed")
  .withColumn("arr", generate_data_udf($"seed"))
  .select(
    $"seed",
    explode($"arr") as "exp"
  )
  .select(
    $"seed",
    $"exp._1" as "n",
    $"exp._2" as "x"
  )

以下是不使用 Row -仅基于rdd。我认为这可能是最有效的方法，因为它使用 map 计算函数输出和 flatMap 为了合并这些输出-这两个操作都在RDD上执行，所以所有的东西都应该被分发。

import numpy as np
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('abc').getOrCreate()
sc = spark.sparkContext

def generate_data(one_integer):
  M = 2 # number of values to generate per seed, e.g. 10M
  np.random.seed(one_integer)
  np_array = np.random.random_sample(M) # generates an array of M random values
  return [(one_integer, i, float(np_array[i])) for i in range(M)]

N = 30 # number of seeds to try, e.g. 100K
list_of_integers = [i for i in range(N)]
list_of_integers_rdd = sc.parallelize(list_of_integers)
generated_data_rdd = list_of_integers_rdd.map(lambda x: generate_data(x))
solved_rdd = generated_data_rdd.flatMap(lambda list: list)

df = spark.createDataFrame(solved_rdd).toDF("seed", "n", "x")
df.show()