dask-spark 0.0.2

示例

pip install dask-spark

从Dask集群创建Spark集群

>>> from dask.distributed import Client
>>> client = Client('scheduler-address:8786')
>>> client
<Client: scheduler='tcp://scheduler-address:8786' processes=8 cores=64>

>>> from dask_spark import dask_to_spark
>>> sc = dask_to_spark(client)
>>> sc
<pyspark.context.SparkContext at 0x7f62fa4bb550>

从Spark集群创建Dask集群

>>> import pyspark
>>> sc = pyspark.SparkContext('local[4]')
<pyspark.context.SparkContext at 0x7f8b908b0128>

>>> from dask_spark import spark_to_dask
>>> client = spark_to_dask(sc)
>>> client
<Client: scheduler="'tcp://127.0.0.1:8786'">

要求和如何工作

这依赖于Dask.distributed的相对较新版本。

对于从Dask启动Spark，假设您已安装Spark，并且 start-master.sh 和 start-slave.sh Spark脚本可用在工作者的PATH上。这将在Dask调度器上启动一个长时间运行的Spark主进程，并在Dask工作者上启动长时间运行的Spark从属进程。每个工作者将只有一个从属。我们将核心数和内存量设置为与Dask工作者和可用内存匹配。

当从Spark启动Dask时，这将阻塞Spark集群。我们在本地机器上启动一个调度器，然后运行一个长时间运行的功能，使用 RDD.mapPartitions 启动一个Dask工作者。

TODO

• [ ] 这很可能会在非平凡情况下失败

• [ ] 启用用户指定Java标志以用于内存和核心使用

• [ ] 支持每个Dask集群多个Spark集群