如果组中存在非障碍，如何删除重复和零？

Question

应根据 flag 列过滤后面的数据框。如果基于列的组 id 和 cod 没有任何 none 的价值的行，则有必要仅维护独特的行，否则，有必要在 flag 中删除无值的行。

import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import col, row_number,max

spark = SparkSession.builder.appName('Vazio').getOrCreate()

data = [('1', 10, 'A'),
('1', 10, 'A'),
('1', 10, None),
('1', 15, 'A'),
('1', 15, None),
('2', 11, 'A'),
('2', 11, 'C'),
('2', 12, 'B'),
('2', 12, 'B'),
('2', 12, 'C'),
('2', 12, 'C'),
('2', 13, None),
('3', 14, None),
('3', 14, None),
('3', 15, None),
('4', 21, 'A'),
('4', 21, 'B'),
('4', 21, 'C'),
('4', 21, 'C')]

df = spark.createDataFrame(data=data, schema = ['id', 'cod','flag'])
df.show()

如何使用Pyspark根据上一个数据框架获取下一个数据帧？

+---+---+----+
| id|cod|flag|
+---+---+----+
|  1| 10|   A|
|  1| 15|   A|
|  2| 11|   A|
|  2| 11|   C|
|  2| 12|   B|
|  2| 12|   C|
|  2| 13|null|
|  3| 14|null|
|  3| 15|null|
|  4| 21|   A|
|  4| 21|   C|
+---+---+----+

Answer 1

这可能是一个解决方案，您可以添加

new_df = pd.DataFrame(columns=df.columns)

for index, row in df.iterrows():
    if row.values.tolist() not in new_df.values.tolist():

        if row["flag"] is None and row.values.tolist()[:2] in new_df[list(new_df.columns)[:2]].values.tolist():
            continue

        new_df.loc[-1] = row.values.tolist()
        new_df.index += 1

最后，

df = new_df.copy(deep=True)
del new_df

Answer 2

使用pyspark。改编自这个答案（spark）

window = Window.partitionBy(['id', 'cod']).orderBy(col('flag').desc())
out = (df.withColumn('row',  row_number().over(window))
         .filter(col('row') == 1).drop('row'))
out.show()

# Output
+---+---+----+
| id|cod|flag|
+---+---+----+
|  2| 11|   A|
|  2| 12|   B|
|  1| 15|   A|
|  3| 14|null|
|  2| 13|null|
|  3| 15|null|
|  1| 10|   A|
+---+---+----+

设置

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import col, row_number

data = [['1', 10, 'A'],
        ['1', 10, 'A'],
        ['1', 10, None],
        ['1', 15, 'A'],
        ['1', 15, None],
        ['2', 11, 'A'],
        ['2', 12, 'B'],
        ['2', 12, 'B'],
        ['2', 13, None],
        ['3', 14, None],
        ['3', 14, None],
        ['3', 15, None]]

columns = ['id', 'cod', 'flag']
spark = SparkSession.builder.getOrCreate()

df = spark.createDataFrame(data = data, schema = columns)
df.show()

# Output
+---+---+----+
| id|cod|flag|
+---+---+----+
|  1| 10|   A|
|  1| 10|   A|
|  1| 10|null|
|  1| 15|   A|
|  1| 15|null|
|  2| 11|   A|
|  2| 12|   B|
|  2| 12|   B|
|  2| 13|null|
|  3| 14|null|
|  3| 14|null|
|  3| 15|null|
+---+---+----+

Answer 3

回答已编辑的问题：

df = df.groupBy('id', 'cod').agg(F.collect_set('flag').alias('flag'))
df = df.withColumn(
    'flag',
    F.explode(F.when(F.size('flag') != 0, F.col('flag')).otherwise(F.array(F.lit(None))))
)
df.show()
# +---+---+----+
# | id|cod|flag|
# +---+---+----+
# |  2| 11|   C|
# |  2| 11|   A|
# |  2| 12|   C|
# |  2| 12|   B|
# |  1| 15|   A|
# |  1| 10|   A|
# |  3| 14|null|
# |  4| 21|   C|
# |  4| 21|   B|
# |  4| 21|   A|
# |  2| 13|null|
# |  3| 15|null|
# +---+---+----+

^{请注意，组（4，21）包含所有3个标志A，B，C。在问题的描述中，您尚未提及仅留下2个值，但是在示例中以某种方式删除了B，因此我将其假定为错误。}

---

回答原始问题：

df = df.groupBy('id', 'cod').agg(F.max('flag').alias('flag'))

您不能仅根据列ID和COD删除重复项，因为无法保证您始终会从中获得值列标志不是 无效的。