【发布时间】:2020-06-20 11:44:29
【问题描述】:
我正在处理一个必须处理大量数据(多个表)的用例,并且我正在尝试将其作为批处理作业提交到 Dataproc 集群 (PySpark)。
我的代码看起来像这样
from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.sql import SparkSession
def readconfig():
#code to read a yaml file
def func(filename, tabname):
sc = SparkContext("local", "First App")
sqlContext = SQLContext(sc)
spark = SparkSession.builder.getOrCreate()
df1= read from file-filename as rdd using sqlcontext
df2= read from bigquery-tabname as df using spark
.
op = intermediate processing
.
#caching and unpersisting 2 dfs
.
op.write.csv(write multiple files in gcs bucket)
sc.stop()
spark.stop()
print("one pair of table and file processed")
if __name__ == "__main__":
config= readconfig()
for i,j in config.items():
func(i,j):
由于文件很大,我正在尝试为正在处理的每对文件和表创建一个单独的SparkSession。它工作正常,我能够处理大量表格。后来我开始收到关于节点内存问题的警告,最后一个错误说:
节点资源不足。无法创建 SparkSession。
为什么在关闭 SparkSession 应该释放上一次迭代的数据内存时会发生这种情况?
【问题讨论】:
标签: python apache-spark pyspark google-cloud-dataproc namenode