小编joc*_*uiz的帖子

我在Ubuntu 15.10中使用Docker容器为Pyspark jupyter/pyspark-notebook工作.我需要安装具有所有依赖关系的folium并在容器中运行Pyspark脚本.我成功安装了Docker,拉出图像并使用命令运行它

docker run -d -p 8888:8888 -p 4040:4040 -v /home/$MYUSER/$MYPROJECT:/home/jovyan/work jupyter/pyspark-notebook

然后,我执行代码示例没有任何问题

import pyspark
sc = pyspark.SparkContext('local[*]')

# do something to prove it works
rdd = sc.parallelize(range(1000))
rdd.takeSample(False, 5)

我查找了conda环境/opt/conda(正如文档中所述),但我的/opt文件夹中没有conda .然后,我将miniconda3和folium安装了所有依赖项作为普通的Python包(不涉及Docker).

它不起作用.当我运行图像并尝试使用import folium它导入包时,找不到folium包:

ImportErrorTraceback (most recent call last)
<ipython-input-1-af6e4f19ef00> in <module>()
----> 1 import folium

ImportError: No module named 'folium'

所以问题可以归结为两个问题:

1. 容器的conda在哪里？
2. 如何将我需要的Python包安装到容器中？

到目前为止,Spark还没有为流数据创建DataFrame,但是当我进行异常检测时,使用DataFrame进行数据分析会更方便,更快捷.我已经完成了这一部分,但是当我尝试使用流数据进行实时异常检测时,出现了问题.我尝试了几种方法仍然无法将DStream转换为DataFrame,也无法将DStream中的RDD转换为DataFrame.

这是我最新版代码的一部分:

import sys
import re

from pyspark import SparkContext
from pyspark.sql.context import SQLContext
from pyspark.sql import Row
from pyspark.streaming import StreamingContext
from pyspark.mllib.clustering import KMeans, KMeansModel, StreamingKMeans
from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark.sql.functions import udf
import operator


sc = SparkContext(appName="test")
ssc = StreamingContext(sc, 5)
sqlContext = SQLContext(sc)

model_inputs = sys.argv[1]

def streamrdd_to_df(srdd):
    sdf = sqlContext.createDataFrame(srdd)
    sdf.show(n=2, truncate=False)
    return sdf

def main():
    indata = ssc.socketTextStream(sys.argv[2], int(sys.argv[3]))
    inrdd = indata.map(lambda r: get_tuple(r))
    Features = Row('rawFeatures')
    features_rdd = …

我有一个Apache Spark集群和一个RabbitMQ代理,我想使用该pyspark.streaming模块消费消息并计算一些指标.

问题是我只找到了这个包,但是用Java和Scala实现.除此之外,我没有在Python中找到任何示例或桥接实现.

我有一个使用Pika实现的消费者,但我不知道如何将有效负载传递给我StreamingContext.