Вы можете попробовать это решение. Сообщите мне, будет ли это быстро.
from pyspark.sql.types import IntegerType, StructField, StructType
values = [
(2,None, None,None,102, 202, 302),
(4,None, None,None,104, 204, 304),
(1,None, None,None,101, 201, 301),
(3,None, None,None,103, 203, 303),
(1,11, 21,31,None,None,None),
(2,12, 22,32,None,None,None),
(4,14, 24,34,None,None,None),
(3,13, 23,33,None,None,None)
]
sc = spark.sparkContext
rdd = sc.parallelize(values)
schema = StructType([
StructField("id", IntegerType(), True),
StructField("col1", IntegerType(), True),
StructField("col2", IntegerType(), True),
StructField("col3", IntegerType(), True),
StructField("sf_1", IntegerType(), True),
StructField("sf_2", IntegerType(), True),
StructField("sf_3", IntegerType(), True)
])
data = spark.createDataFrame(rdd, schema)
data.show()
# +---+----+----+----+----+----+----+
# | id|col1|col2|col3|sf_1|sf_2|sf_3|
# +---+----+----+----+----+----+----+
# | 2|null|null|null| 102| 202| 302|
# | 4|null|null|null| 104| 204| 304|
# | 1|null|null|null| 101| 201| 301|
# | 3|null|null|null| 103| 203| 303|
# | 1| 11| 21| 31|null|null|null|
# | 2| 12| 22| 32|null|null|null|
# | 4| 14| 24| 34|null|null|null|
# | 3| 13| 23| 33|null|null|null|
# +---+----+----+----+----+----+----+
data.createOrReplaceTempView("data")
join_key = 'id'
table_name = 'data'
query = "{0}".format(join_key)
filters = ""
for index, column_name in enumerate(data.columns):
if join_key != column_name:
query += ",\n\t case when a." + column_name + " is null then b." + column_name + " else a." + column_name + " end as " + column_name
filters += "\nAND {0} IS NOT NULL".format(column_name) if index !=1 else " {0} IS NOT NULL".format(column_name)
final_query ="""
SELECT a.{1}
FROM {0} a INNER JOIN {0} b ON a.{2} = b.{2}
""".format(table_name, query, join_key)
print(final_query)
# SELECT a.id,
# case when a.col1 is null then b.col1 else a.col1 end as col1,
# case when a.col2 is null then b.col2 else a.col2 end as col2,
# case when a.col3 is null then b.col3 else a.col3 end as col3,
# case when a.sf_1 is null then b.sf_1 else a.sf_1 end as sf_1,
# case when a.sf_2 is null then b.sf_2 else a.sf_2 end as sf_2,
# case when a.sf_3 is null then b.sf_3 else a.sf_3 end as sf_3
# FROM data a INNER JOIN data b ON a.id = b.id
print(filters)
# col1 IS NOT NULL
# AND col2 IS NOT NULL
# AND col3 IS NOT NULL
# AND sf_1 IS NOT NULL
# AND sf_2 IS NOT NULL
# AND sf_3 IS NOT NULL
spark.sql(final_query).dropDuplicates().filter(filters).show()
# +---+----+----+----+----+----+----+
# | id|col1|col2|col3|sf_1|sf_2|sf_3|
# +---+----+----+----+----+----+----+
# | 1| 11| 21| 31| 101| 201| 301|
# | 3| 13| 23| 33| 103| 203| 303|
# | 4| 14| 24| 34| 104| 204| 304|
# | 2| 12| 22| 32| 102| 202| 302|
# +---+----+----+----+----+----+----+