rdd.foreach pysparkに関数を適用する

Question

rddの各行にsend_active_mqを適用しようとしています。

def send_to_active_mq(json_string) : 
    k = str(json_string) 
    conn.send(body=k,destination='dwEmailsQueue2') 

json_rdd_to_send.foreach(send_to_active_mq) 

私は火花のドキュメント http://spark.apache.org/docs/1.1.1/api/python/pyspark.rdd.RDD-class.html#foreach

で提案されているのと同じ方法を適用しかし、私は次のエラーを取得しています。

AttributeError: 'builtin_function_or_method' object has no attribute '__code__' 

ここでは何か根本的に間違っていますか？

Answer 1

これはおそらくconnオブジェクトに接続されています。お試しください：

def send_to_active_mq(json_strings): 
    conn = ... # Initalize connection 
    for json_string in json_strings: 
     conn.send(body=str(json_string) ,destination='dwEmailsQueue2') 

json_rdd_to_send.foreachPartition(send_to_active_mq)