相关疑难解决方法(0)

我最近在我的服务器上使用PySpark与Ipython一起使用24个CPU和32GB RAM.它只能在一台机器上运行.在我的过程中,我想收集大量数据,如下面的代码所示:

train_dataRDD = (train.map(lambda x:getTagsAndText(x))
.filter(lambda x:x[-1]!=[])
.flatMap(lambda (x,text,tags): [(tag,(x,text)) for tag in tags])
.groupByKey()
.mapValues(list))

当我做

training_data =  train_dataRDD.collectAsMap()

它给了我outOfMemory错误.Java heap Space.此外,我在此错误后无法对Spark执行任何操作,因为它失去了与Java的连接.它给出了Py4JNetworkError: Cannot connect to the java server.

看起来堆空间很小.如何将其设置为更大的限制？

编辑:

我在跑步之前尝试过的事情: sc._conf.set('spark.executor.memory','32g').set('spark.driver.memory','32g').set('spark.driver.maxResultsSize','0')

我按照此处的文档更改了spark选项(如果你执行ctrl-f并搜索spark.executor.extraJavaOptions):http://spark.apache.org/docs/1.2.1/configuration.html

它说我可以通过设置spark.executor.memory选项来避免OOM.我做了同样的事情,但似乎没有工作.

上下文： 集群配置如下：

• 一切都与docker文件一起运行。
• 节点1：Spark Master
• node2：jupyter中心（我也在其中运行笔记本）
• 节点3-7：Spark Worker节点
• 我可以使用spark的默认端口从我的工作节点telnet并ping通到node2，反之亦然

问题： 我试图在以集群部署模式运行的pyspark jupyter笔记本中创建一个Spark会话。我试图使驱动程序在不是运行jupyter笔记本的节点上运行。现在，我可以在群集上运行作业，但只能使用在node2上运行的驱动程序。

经过大量的挖掘，我发现了这个stackoverflow帖子，其中声称如果您使用spark运行交互式shell，则只能在本地部署模式（驱动程序位于您正在使用的计算机上）中进行。该帖子继续说，类似jupyter hub之类的结果也无法在集群部署模式下工作，但是我找不到任何可以证实这一点的文档。有人可以确认jupyter hub是否可以完全在集群模式下运行吗？

我尝试以集群部署模式运行spark会话：

from pyspark.sql import SparkSession

spark = SparkSession.builder\
.enableHiveSupport()\
.config("spark.local.ip",<node 3 ip>)\
.config("spark.driver.host",<node 3 ip>)\
.config('spark.submit.deployMode','cluster')\
.getOrCreate()

错误：

/usr/spark/python/pyspark/sql/session.py in getOrCreate(self)
    167                     for key, value in self._options.items():
    168                         sparkConf.set(key, value)
--> 169                     sc = SparkContext.getOrCreate(sparkConf)
    170                     # This SparkContext may be an existing one.
    171                     for key, value in self._options.items():

/usr/spark/python/pyspark/context.py in getOrCreate(cls, conf)
    308         with SparkContext._lock:
    309             if SparkContext._active_spark_context is None:
--> …

我正在尝试将函数应用于pyspark中的每个数据集组.我得到的第一个错误是

Py4JError: An error occurred while calling o62.__getnewargs__. Trace:
py4j.Py4JException: Method __getnewargs__([]) does not exist

Run Code Online (Sandbox Code Playgroud)

为了解决上述问题,我删除了火花功能(我有spark.range()).现在错误已经解决,但现在我得到以下内容:

File "/opt/cloudera/parcels/SPARK2-2.3.0.cloudera2-1.cdh5.13.3.p0.316101/lib/spark2/python/pyspark/serializers.py", line 276, in load_stream
    import pyarrow as pa
ImportError: No module named pyarrow 

Run Code Online (Sandbox Code Playgroud)

但是当我自己尝试它时,它就可以了.

df = pd.DataFrame({"a": [1, 2, 3]})
pa.Table.from_pandas(df)
pyarrow.Table
a: int64
__index_level_0__: int64
metadata
--------
{'pandas': '{"pandas_version": "0.23.0", "index_columns": ["__index_level_0__"], "columns": [{"metadata": null, "field_name": "a", "name": "a", "numpy_type": "int64", "pandas_type": "int64"}, {"metadata": null, "field_name": "__index_level_0__", "name": null, "numpy_type": "int64", "pandas_type": "int64"}], "column_indexes": [{"metadata": null, "field_name": null, "name": …