如何从工作人员访问SparkSession？

Question

我在databricks中使用spark结构化流媒体。在这里，我使用foreach操作对每个数据记录执行一些操作。但是我传递给foreach的函数使用了SparkSession，但它抛出了一个错误：_pickle。PicklingError:无法序列化对象：异常：似乎您正试图从广播变量、操作或转换引用SparkContext。SparkContext只能在驱动程序上使用，不能在工作程序上运行的代码中使用。

那么，有没有办法在foreach中使用SparkSession？

编辑#1:foreach中传递的函数示例如下：

def process_data(row):
  df = spark.createDataFrame([row])    
  df.write.mode("overwrite").saveAsTable("T2")
  spark.sql("""
    MERGE INTO T1
    USING T2 ON T2.type="OrderReplace" AND T1.ReferenceNumber=T2.originalReferenceNumber
    WHEN MATCHED THEN
    UPDATE SET
      shares = T2.shares,
      price = T2.price,
      ReferenceNumber = T2.newReferenceNumber
  """)

所以，我需要Spark会话在这里，这是不可用的Foreach。

Answer 1

通过对要在流数据集上逐行执行的任务的描述，可以尝试使用foreachBatch。

流媒体数据集可能有数千到数百万条记录，如果您在每条记录上访问外部系统，这将是一个巨大的开销。从您的示例中，您似乎正在根据T2中的事件更新基本数据集T1。

通过上面的链接，您可以执行以下操作：，

def process_data(df, epoch_id):    
  df.write.mode("overwrite").saveAsTable("T2")
  spark.sql("""
    MERGE INTO T1
    USING T2 ON T2.type="OrderReplace" AND T1.ReferenceNumber=T2.originalReferenceNumber
    WHEN MATCHED THEN
    UPDATE SET
      shares = T2.shares,
      price = T2.price,
      ReferenceNumber = T2.newReferenceNumber
  """)

streamingDF.writeStream.foreachBatch(process_data).start()   

关于你正在做的事情，还有一些额外的要点，

1. 您正在尝试对流数据集执行foreach并覆盖T2hive表。由于记录是在集群管理器（如warn）中并行处理的，因此可能存在损坏的T2，因为多个任务可能会更新它