我正在编写一个用户定义的函数,它将获取除数据帧中第一个之外的所有列并进行求和(或任何其他操作).现在数据框有时可以有3列或4列或更多列.它会有所不同.
我知道我可以硬编码4个列名作为UDF传递,但在这种情况下它会有所不同所以我想知道如何完成它?
以下是第一个示例中的两个示例,我们有两列要添加,第二个示例中我们有三列要添加.
我有一个具有以下结构的Spark数据帧.bodyText_token具有标记(处理/单词集).我有一个已定义关键字的嵌套列表
root
|-- id: string (nullable = true)
|-- body: string (nullable = true)
|-- bodyText_token: array (nullable = true)
keyword_list=['union','workers','strike','pay','rally','free','immigration',],
['farmer','plants','fruits','workers'],['outside','field','party','clothes','fashions']]
Run Code Online (Sandbox Code Playgroud)
我需要检查每个关键字列表下有多少令牌,并将结果添加为现有数据帧的新列.例如:如果tokens =["become", "farmer","rally","workers","student"]
结果是 - > [1,2,0]
以下功能按预期工作.
def label_maker_topic(tokens,topic_words):
twt_list = []
for i in range(0, len(topic_words)):
count = 0
#print(topic_words[i])
for tkn in tokens:
if tkn in topic_words[i]:
count += 1
twt_list.append(count)
return twt_list
Run Code Online (Sandbox Code Playgroud)
我在withColumn下使用了udf来访问该函数,但是我收到了一个错误.我认为这是关于将外部列表传递给udf.有没有办法可以将外部列表和datafram列传递给udf并向我的数据帧添加新列?
topicWord = udf(label_maker_topic,StringType())
myDF=myDF.withColumn("topic_word_count",topicWord(myDF.bodyText_token,keyword_list))
Run Code Online (Sandbox Code Playgroud) python user-defined-functions apache-spark apache-spark-sql pyspark