在 PySpark 中的文件之间传递 spark 上下文作为参数

Question

相关代码片段：

文件 1：master.py

# Spark Imports
from pyspark import SparkContext,SparkConf
from pyspark.streaming import StreamingContext
from pyspark.sql import SQLContext

#Import self defined function
from helper import enrichment


def ingestion(sc,ssc):
    # Work with stream
    kafkaStream = KafkaUtils.createStream(ssc, zkQuorum, "streaming-consumer", {topic: 1})
    # Call function defined in helper.py
    enriched_data = kafkaStream_json.map(lambda single_log:enrichment(single_log,client_id,machine_id))

if __name__ == "__main__":
    # Name of Spark App
    conf = SparkConf().setAppName("Test")

    # Spark and Spark streaming configuration
    sc = SparkContext(conf=conf)
    ssc = StreamingContext(sc, 1)
    ingestion(sc,ssc)
    # Start the stream and keep it running unless terminated
    ssc.start()
    ssc.awaitTermination()

文件 2：helper.py

from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext
def enrichment():
    test_df = pd.DataFrame(some operations...)
    spark_df = sqlContext.createDataFrame(test_df)
    ...

面临的问题：

其中的流媒体部分工作正常，但是当我调用函数 enrichment 时，根据使用情况，我面临以下问题：

Case1：运行上面的例子时，它说：

spark_df = sqlContext.createDataFrame(test_df)
NameError: global name 'sqlContext' is not defined

案例 2：当我将 spark Context 作为参数传递时，显示的消息如下：

“例外：您似乎正在尝试引用 来自广播变量、动作或转换的 SparkContext。 SparkContext 只能在驱动程序上使用，不能在它运行的代码中使用 对工人。有关详细信息，请参阅 SPARK-5063。”

这是我找到的最接近的解决方案： ERROR:SparkContext can only be used on the driver, not in code that it run on workers. For more information, see SPARK-5063

但是，它似乎无法解决我的问题。任何线索将不胜感激。

我需要将它们作为两个单独的文件，内联不起作用。代码运行使用：

sudo $SPARK_HOME/spark-submit --master local[2] /home/user/master.py

Answer 1

我认为你应该使用 SparkSession。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('abc').getOrCreate()

您可以将 spark 作为参数传递给丰富函数：

def enrichment(spark):
    test_df = pd.DataFrame(some operations...)
    spark_df = spark.createDataFrame(test_df)
    ...

或：

def enrichment():
    spark = SparkSession.builder.getOrCreate()
    test_df = pd.DataFrame(some operations...)
    spark_df = spark.createDataFrame(test_df)
    ...