python之在 PySpark 中使用 toDF() 函数将 RDD 转换为 Dataframe 时的奇怪行为

我是 Spark 的新手。当我使用 toDF() 函数将 RDD 转换为数据帧时，它似乎计算了我之前编写的所有转换函数，如 map() 。我想知道 PySpark 中的 toDF() 是转换还是 Action 。

我创建了一个简单的 RDD 并使用一个简单的函数输出它的值，仅用于测试，并在 map() 之后使用 toDF()。结果似乎部分运行了 map 中的功能。当我显示数据帧的结果时，toDF() 就像转换一样再次输出结果。

>>> a = sc.parallelize([(1,),(2,),(3,)]) 
>>> def f(x): 
...     print(x[0]) 
...     return (x[0] + 1, ) 
... 
>>> b = a.map(f).toDF(["id"]) 
2 
1 
>>> b = a.map(f).toDF(["id"]).show() 
2 
1 
1 
2 
3 
+---+ 
| id| 
+---+ 
|  2| 
|  3| 
|  4| 
+---+ 

有人能告诉我为什么 PySpark 中的 toDF() 函数既像 Action 又像转换吗？非常感谢。

PS:在 Scala 中，toDF 在我的例子中就像转换。

请您参考如下方法:

这并不奇怪。由于您没有提供架构，Spark 必须根据数据进行推断。如果 RDD 是输入，它将调用 SparkSession._createFromRDD随后 SparkSession._inferSchema ，如果缺少 samplingRatio，will evaluate up to 100 row :

first = rdd.first() 
if not first: 
    raise ValueError("The first row in RDD is empty, " 
                     "can not infer schema") 
if type(first) is dict: 
    warnings.warn("Using RDD of dict to inferSchema is deprecated. " 
                  "Use pyspark.sql.Row instead") 
 
 
if samplingRatio is None: 
    schema = _infer_schema(first, names=names) 
    if _has_nulltype(schema): 
        for row in rdd.take(100)[1:]: 
            schema = _merge_type(schema, _infer_schema(row, names=names)) 
            if not _has_nulltype(schema): 
                break 
        else: 
            raise ValueError("Some of types cannot be determined by the " 
                             "first 100 rows, please try again with sampling") 

现在剩下的唯一谜题是为什么它不准确地评估一条记录。毕竟在你的情况下 first 不是空的并且不包含 None。

那是因为 first 是通过 take 实现的，并且不保证会评估确切数量的项目。如果第一个分区没有产生所需数量的项目，它将迭代地增加要扫描的分区数量。请查看the implementation了解详情。

如果您想避免这种情况，您应该使用 createDataFrame 并以 DDL 字符串形式提供架构:

spark.createDataFrame(a.map(f), "val: integer") 

或等效的StructType。

你不会在 Scala 对应物中找到任何类似的行为，因为它不在 toDF 中使用模式推断。它要么从 Encoder(使用 Scala 反射获取)中检索相应的模式，要么根本不允许转换。最接近的相似行为是对输入源的推断，如 CSV or JSON :

spark.read.json(Seq("""{"foo": "bar"}""").toDS.map(x => { println(x); x }))