Estou tentando usarSQLContext.subtract () no Spark 1.6.1 para remover linhas de um quadro de dados com base em uma coluna de outro quadro de dados. Vamos usar um exemplo:

from pyspark.sql import Row

df1 = sqlContext.createDataFrame([
    Row(name='Alice', age=2),
    Row(name='Bob', age=1),
]).alias('df1')

df2 = sqlContext.createDataFrame([
    Row(name='Bob'),
])

df1_with_df2 = df1.join(df2, 'name').select('df1.*')
df1_without_df2 = df1.subtract(df1_with_df2)

Desde que eu quero todas as linhas dedf1 que não incluemname='Bob' eu esperoRow(age=2, name='Alice'). Mas eu também recupero Bob:

print(df1_without_df2.collect())
# [Row(age='1', name='Bob'), Row(age='2', name='Alice')]

Após várias experiências para chegar a esteMCVE, Descobri que o problema está noage chave. Se eu omitir:

df1_noage = sqlContext.createDataFrame([
    Row(name='Alice'),
    Row(name='Bob'),
]).alias('df1_noage')

df1_noage_with_df2 = df1_noage.join(df2, 'name').select('df1_noage.*')
df1_noage_without_df2 = df1_noage.subtract(df1_noage_with_df2)
print(df1_noage_without_df2.collect())
# [Row(name='Alice')]

Então eu só pego Alice como esperado. A observação mais estranha que fiz é que é possível adicionar chaves, desde que sejamdepois de (no sentido da ordem lexicográfica) a chave que eu uso na junção:

df1_zage = sqlContext.createDataFrame([
    Row(zage=2, name='Alice'),
    Row(zage=1, name='Bob'),
]).alias('df1_zage')

df1_zage_with_df2 = df1_zage.join(df2, 'name').select('df1_zage.*')
df1_zage_without_df2 = df1_zage.subtract(df1_zage_with_df2)
print(df1_zage_without_df2.collect())
# [Row(name='Alice', zage=2)]

Eu entendo corretamente Alice (com seu zage)! Nos meus exemplos reais, estou interessado em todas as colunas, não apenas nas que estão atrásname.