使用 Pyspark，我如何选择/保留包含非空值的 DataFrame 的所有 列；或等效地删除所有不包含数据的列。
已编辑：根据 Suresh 要求，
for column in media.columns:
if media.select(media[column]).distinct().count() == 1:
media = media.drop(media[column])
这里我假设如果 count 是一，那么它应该是 Nan。但我想检查那是不是南。如果还有其他内置 spark 功能，请告诉我。
【问题讨论】：
标签： python apache-spark pyspark
我试过了。说，我有一个如下的数据框，
from pyspark.sql import functions as F
>>> df.show()
+----+----+----+
|col1|col2|col3|
+----+----+----+
| 1| 2|null|
|null| 3|null|
| 5|null|null|
+----+----+----+
>>> df1 = df.agg(*[F.count(c).alias(c) for c in df.columns])
>>> df1.show()
+----+----+----+
|col1|col2|col3|
+----+----+----+
| 2| 2| 0|
+----+----+----+
>>> nonNull_cols = [c for c in df1.columns if df1[[c]].first()[c] > 0]
>>> df = df.select(*nonNull_cols)
>>> df.show()
+----+----+
|col1|col2|
+----+----+
| 1| 2|
|null| 3|
| 5|null|
+----+----+
【讨论】：
F...？
对我来说，它的工作方式与@Suresh 答案有点不同：
nonNull_cols = [c for c in original_df.columns if original_df.filter(func.col(c).isNotNull()).count() > 0]
new_df = original_df.select(*nonNull_cols)
【讨论】：
这是一个更有效的解决方案，它不涉及对列进行循环。当您有很多列时，它会快得多。我在一个有 800 列的数据帧上测试了这里的其他方法，运行时间为 17 分钟。在我对同一数据集的测试中，以下方法仅需 1 分钟。
def drop_fully_null_columns(df, but_keep_these=[]):
"""Drops DataFrame columns that are fully null
(i.e. the maximum value is null)
Arguments:
df {spark DataFrame} -- spark dataframe
but_keep_these {list} -- list of columns to keep without checking for nulls
Returns:
spark DataFrame -- dataframe with fully null columns removed
"""
# skip checking some columns
cols_to_check = [col for col in df.columns if col not in but_keep_these]
if len(cols_to_check) > 0:
# drop columns for which the max is None
rows_with_data = df.select(*cols_to_check).groupby().agg(*[F.max(c).alias(c) for c in cols_to_check]).take(1)[0]
cols_to_drop = [c for c, const in rows_with_data.asDict().items() if const == None]
new_df = df.drop(*cols_to_drop)
return new_df
else:
return df
【讨论】：
这样做的间接方法之一是
import pyspark.sql.functions as func
for col in sdf.columns:
if (sdf.filter(func.isnan(func.col(col)) == True).count() == sdf.select(func.col(col)).count()):
sdf = sdf.drop(col)
更新：
上面的代码删除了所有 nan 的列。如果您正在寻找所有空值，那么
import pyspark.sql.functions as func
for col in sdf.columns:
if (sdf.filter(func.col(col).isNull()).count() == sdf.select(func.col(col)).count()):
sdf = sdf.drop(col)
如果我找到一些最佳方法，我会更新我的答案:-)
【讨论】：
这是我的管道中用于删除空列的功能。 希望对您有所帮助！
# Function to drop the empty columns of a DF
def dropNullColumns(df):
# A set of all the null values you can encounter
null_set = {"none", "null" , "nan"}
# Iterate over each column in the DF
for col in df.columns:
# Get the distinct values of the column
unique_val = df.select(col).distinct().collect()[0][0]
# See whether the unique value is only none/nan or null
if str(unique_val).lower() in null_set:
print("Dropping " + col + " because of all null values.")
df = df.drop(col)
return(df)
【讨论】：
data_2 = { 'furniture': [np.NaN ,np.NaN ,True], 'myid': ['1-12', '0-11', '2-12'], 'clothing': ["pants", "shoes", "socks"]} df_1 = pd.DataFrame(data_2) ddf_1 = spark.createDataFrame(df_1)你会看到家具栏会被丢弃，虽然实际上它不应该被丢弃。
或者只是
from pyspark.sql.functions import col
for c in df.columns:
if df.filter(col(c).isNotNull()).count() == 0:
df = df.drop(c)
【讨论】：
这是一个强大的解决方案，它考虑了列中所有可能的空值组合。首先，找到所有空列，然后将它们删除。它看起来冗长而繁琐，但实际上这是一个强大的解决方案。仅使用一个循环来查找空列，并且没有应用诸如 collect() 之类的内存密集型函数，这应该使该解决方案快速高效。
rows = [(None, 18, None, None),
(1, None, None, None),
(1, 9, 4.0, None),
(None, 0, 0., None)]
schema = "a: int, b: int, c: float, d:int"
df = spark.createDataFrame(data=rows, schema=schema)
def get_null_column_names(df):
column_names = []
for col_name in df.columns:
min_ = df.select(F.min(col_name)).first()[0]
max_ = df.select(F.max(col_name)).first()[0]
if min_ is None and max_ is None:
column_names.append(col_name)
return column_names
null_columns = get_null_column_names(df)
def drop_column(null_columns, df):
for column_ in null_columns:
df = df.drop(column_)
return df
df = drop_column(null_columns, df)
df.show()
【讨论】：
只是从上面的答案中挑选出来，为我的用例编写了我自己的解决方案。
我实际上想要做的是从我的 pyspark 数据框中删除所有具有 100% 空值的列。
# identify and remove all columns having 100% null values
df_summary_count = your_df.summary("count")
null_cols = [c for c in df_summary_count .columns if df_summary_count.select(c).first()[c] == '0']
filtered_df = df_summary_count .drop(*null_cols)
【讨论】：
创建一个数据框：
df = spark.createDataFrame(
[
(1, 'baz'),
(2, 'baz')
],
['foo','bar']
)
添加一个空列：
df = df.withColumn('foobar', lit(None))
制作非空列的列表：
non_null_columns = df.summary('count').drop('summary').columns
创建df 中也存在于non_null_columns 中的列的列表理解，并使用这些列从您的df 中进行选择：
df.select([col for col in df.columns if col in non_null_columns]).show()
哪个打印：
+---+---+
|foo|bar|
+---+---+
| 1|baz|
| 2|baz|
+---+---+
【讨论】：