使用带参数的PySpark 3 DataFrame#transform方法
使用带参数的PySpark 3 DataFrame#transform方法
提问于 2020-06-06 14:31:35
这个问题讨论了如何链接自定义PySpark 2转换。
DataFrame#transform法被添加到PySpark 3 API中。
此代码片段显示了不接受参数并按预期工作的自定义转换，以及另一个接受参数但不起作用的自定义转换。
from pyspark.sql.functions import col, lit
df = spark.createDataFrame([(1, 1.0), (2, 2.)], ["int", "float"])
def with_funny(word):
def inner(df):
return df.withColumn("funny", lit(word))
return inner
def cast_all_to_int(input_df):
return input_df.select([col(col_name).cast("int") for col_name in input_df.columns])
df.transform(with_funny("bumfuzzle")).transform(cast_all_to_int).show()下面是输出的内容：
+---+-----+-----+
|int|float|funny|
+---+-----+-----+
| 1| 1| null|
| 2| 2| null|
+---+-----+-----+如何定义with_funny()方法以输出PySpark 3 API的值？
回答 2
Stack Overflow用户
回答已采纳
发布于 2020-08-27 08:12:08
如果我理解，您的第一个transform方法将添加一个新列，其文字作为参数传递，最后一个转换将所有列强制转换为int类型，对吗？
将字符串转换为int将返回一个空值，您的最终输出是正确的：
from pyspark.sql.functions import col, lit
df = spark.createDataFrame([(1, 1.0), (2, 2.)], ["int", "float"])
def with_funny(word):
def inner(df):
return df.withColumn("funny", lit(word))
return inner
def cast_all_to_int(input_df):
return input_df.select([col(col_name).cast("int") for col_name in input_df.columns])
#first transform
df1 = df.transform(with_funny("bumfuzzle"))
df1.show()
#second transform
df2 = df1.transform(cast_all_to_int)
df2.show()
#all together
df_final = df.transform(with_funny("bumfuzzle")).transform(cast_all_to_int)
df_final.show()输出：
+---+-----+---------+
|int|float| funny|
+---+-----+---------+
| 1| 1.0|bumfuzzle|
| 2| 2.0|bumfuzzle|
+---+-----+---------+
+---+-----+-----+
|int|float|funny|
+---+-----+-----+
| 1| 1| null|
| 2| 2| null|
+---+-----+-----+
+---+-----+-----+
|int|float|funny|
+---+-----+-----+
| 1| 1| null|
| 2| 2| null|
+---+-----+-----+也许您想要的是像这样切换转换的顺序：
df_final = df.transform(cast_all_to_int).transform(with_funny("bumfuzzle"))
df_final.show()输出：
+---+-----+---------+
|int|float| funny|
+---+-----+---------+
| 1| 1|bumfuzzle|
| 2| 2|bumfuzzle|
+---+-----+---------+Stack Overflow用户
发布于 2022-09-06 11:56:05
这一问题已在火星雨3.3.0中得到解决。
def transform(self, func: Callable[..., "DataFrame"], *args: Any, **kwargs: Any) -> "DataFrame":
"""Returns a new :class:`DataFrame`. Concise syntax for chaining custom transformations.
.. versionadded:: 3.0.0
Parameters
----------
func : function
a function that takes and returns a :class:`DataFrame`.
*args
Positional arguments to pass to func.
.. versionadded:: 3.3.0
**kwargs
Keyword arguments to pass to func.
.. versionadded:: 3.3.0
Examples
--------
>>> from pyspark.sql.functions import col
>>> df = spark.createDataFrame([(1, 1.0), (2, 2.0)], ["int", "float"])
>>> def cast_all_to_int(input_df):
... return input_df.select([col(col_name).cast("int") for col_name in input_df.columns])
>>> def sort_columns_asc(input_df):
... return input_df.select(*sorted(input_df.columns))
>>> df.transform(cast_all_to_int).transform(sort_columns_asc).show()
+-----+---+
|float|int|
+-----+---+
| 1| 1|
| 2| 2|
+-----+---+
>>> def add_n(input_df, n):
... return input_df.select([(col(col_name) + n).alias(col_name)
... for col_name in input_df.columns])
>>> df.transform(add_n, 1).transform(add_n, n=10).show()
+---+-----+
|int|float|
+---+-----+
| 12| 12.0|
| 13| 13.0|
+---+-----+
"""
result = func(self, *args, **kwargs)
assert isinstance(
result, DataFrame
), "Func returned an instance of type [%s], " "should have been DataFrame." % type(result)
return result页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/62233150
复制相关文章
相似问题
腾讯云开发者
