散景服务器允许用户在回叫时执行几乎任何python代码。
我想知道它是否也可用于运行Spark作业。
到目前为止,我在这里找到了一些想法(Best Practice to launch Spark Applications via Web Application?),但我不确定。
为了使它更具体一点:
我知道这个帖子超级老了,但最近我有完全相同的问题。
我在我的散景应用程序中运行了Spark。我所做的不是生产级部署,但它确实有效,让人们自我服务。需要注意的一些事情让它对我有用:
这是我的散景服务器main.py
的简化外观(这是开源的,你可以在这里看到 - https://github.com/mozilla/overscripted-explorer/blob/22feeedaf655bd7058331a5217900b0d2f41448b/text_search/main.py)
实例化火花。 getOrCreate
在这里很重要:
from pyspark import SparkContext, SQLContext
sc = SparkContext.getOrCreate()
spark = SQLContext(sc)
....
def do_spark_computation():
....
df = spark.read.parquet(DATA_FILE)
frac = sample_frac.value / 100 # sample_frac is a bokeh widget
sample = df.sample(False, frac)
....
....
对于非阻塞,我从散景文档:https://bokeh.pydata.org/en/latest/docs/user_guide/server.html#updating-from-unlocked-callbacks这个例子中抄下来
from concurrent.futures import ThreadPoolExecutor
from functools import partial
from bokeh.document import without_document_lock
from bokeh.io import curdoc
from tornado.gen import coroutine
EXECUTOR = ThreadPoolExecutor(max_workers=2)
doc = curdoc() # It was important to set this up globally
def do_spark_computation():
....
df = spark.read.parquet(DATA_FILE)
frac = sample_frac.value / 100 # sample_frac is a bokeh widget
sample = df.sample(False, frac)
....
@coroutine
@without_document_lock
def get_new_data():
doc.add_next_tick_callback(function_updates_bokeh_models)
results = yield EXECUTOR.submit(do_spark_computation)
doc.add_next_tick_callback(partial(function_updates_bokeh_models, results))
apply_button.on_click(get_new_data)