迭代列表的每个元素并传递给函数

Posted 2021-05-01

我想迭代列表中的每个元素并将其传递给函数。这是我尝试但低于错误。

import call_functions
newObject = call_functions.call_functions_class()
size_DF = newObject.descr_size(sc, dataBase)
size_RDD = sc.parallelize(size_DF).map(lambda x : x[0])

def full_item_new(sc, dataBase, length, end_date):
    newObject.full_item(sc, dataBase, length, end_date)
size_RDD.map(lambda x : full_item_new(sc, dataBase, x[0], end_date)).collect()

def full_item(sc, dataBase, length, end_date):
    sqlContext = SQLContext(sc)
    insertDF = sqlContext.sql("insert into -----")
    return insertDF

错误：

“看起来你正试图从广播引用SparkContext”例外：你似乎试图从广播变量，动作或转换中引用SparkContext。 SparkContext只能在驱动程序上使用，而不能在工作程序上运行的代码中使用。有关更多信息，请参阅SPARK-5063。

答案

没有看到函数full_item_new，很难回答，但只是看看args，你提供sc，这显然是你的火花上下文变量。因此，这意味着当您在rdd上执行转换时，您正尝试使用sc执行操作或转换。这根本不可能。

full_item_new是对工人执行的，但sc只能用于驱动程序。