内容

在 pySpark 中，我使用以下代码向所有节点广播一个变量：

sc = spark.sparkContext # Get context

# Extract stopwords from a file in hdfs
# The result looks like stopwords = {"and", "foo", "bar" ... }
stopwords = set([line[0] for line in csv.reader(open(SparkFiles.get("stopwords.txt"), 'r'))])

# The set of stopwords is broadcasted now
stopwords = sc.broadcast(stopwords)

在广播stopwords之后，我想让它在mapPartitions中可访问：

# Some dummy-dataframe
df = spark.createDataFrame([(["TESTA and TESTB"], ), (["TESTB and TESTA"], )], ["text"])

# The method which will be applied to mapPartitions
def stopwordRemoval(partition, passed_broadcast):
    """
    Removes stopwords from "text"-column.

    @partition: iterator-object of partition.
    @passed_stopwords: Lookup-table for stopwords.
    """

    # Now the broadcast is passed
    passed_stopwords = passed_broadcast.value

    for row in partition:
        yield [" ".join((word for word in row["text"].split(" ") if word not in passed_stopwords))]

# re-partitioning in order to get mapPartitions working
df = df.repartition(2)

# Now apply the method
df = df.select("text").rdd \
        .mapPartitions(lambda partition: stopwordRemoval(partition, stopwords)) \
        .toDF()

# Result
df.show()

#Result:
+------------+
| text       |
+------------+
|TESTA TESTB |
|TESTB TESTA |
+------------+

问题

尽管它能工作，但我不太确定这是否是广播变量的正确用法。因此，我的问题是：
1.当我以演示的方式将广播传递给mapParitions时，是否正确执行了广播？
1.在mapParitions中使用广播是否有用，因为无论如何，停用词都将随函数分发到所有节点（停用词从不重用）？
第二个问题与this question有关，它部分地回答了我自己的问题。这就是为什么我也选择问这个问题。