python - 无法序列化org.apache.hadoop.io.DoubleWritable类-MongoDB Hadoop Connector + Spark + Python

Question

遵循了这个好的blog post之后，我找到了他代码的github repo，并对其进行了cloned设置，以使其易于他人使用。

我创建了一个script来:

下载并构建spark。

下载，应用补丁并构建mongodb hadoop connector。

下载mongodb java driver v3.1.1 jar。

将最少的 jar 放在一起

下载sample file

将其导入到mongodb集合

安装一个必要的python库pytz

它准备一切准备运行。

python script基本上是这样的:

config = {"mongo.input.uri": "mongodb://localhost:27017/marketdata.minbars"}
inputFormatClassName = "com.mongodb.hadoop.MongoInputFormat"
keyClassName = "org.apache.hadoop.io.Text"
valueClassName = "org.apache.hadoop.io.MapWritable"

minBarRawRDD = sc.newAPIHadoopRDD(inputFormatClassName, keyClassName, valueClassName, None, None, config)
minBarRDD = minBarRawRDD.values()

import calendar, time, math
dateFormatString = '%Y-%m-%d %H:%M'     
groupedBars = minBarRDD.sortBy(lambda doc: str(doc["Timestamp"])).groupBy(lambda doc: 
    (doc["Symbol"], math.floor(calendar.timegm(time.strptime(doc["Timestamp"], dateFormatString)) / (5*60))))

def ohlc(grouping):
    # some

config["mongo.output.uri"] = "mongodb://localhost:27017/marketdata.fiveminutebars"
outputFormatClassName = "com.mongodb.hadoop.MongoOutputFormat"
# resultRDD.saveAsNewAPIHadoopFile("file:///placeholder", outputFormatClassName, None, None, None, None, config)

最后一行注释了当我运行 spark-ohlcbars-example.submit.sh脚本时，一切运行正常，没有任何错误。

但是一旦取消注释onder的最后一行以尝试将数据保存回mongodb中，就会引发异常

Can't serialize class org.apache.hadoop.io.DoubleWritable

我猜想从Python> Java> MongoDB格式转换的过程会在尝试序列化时丢失，并且无法正常工作。

您可以在 issue中看到 git repository

我希望社区有办法解决我们的问题。你们知道如何克服这个问题吗？

Answer 1

通过在提交哈希 affad1b7 上使用pymongo_spark可以解决该特定问题。

我将文件复制到我的项目中，并在主要的python脚本上添加了3行代码:

import pymongo_spark
pymongo_spark.activate()
...
# at the end of the script
resultRDD.saveToMongoDB(config["mongo.output.uri"])

您可以在github commit上看到完整的差异

https://github.com/danielsan/mongodb-analytics-examples/commit/f287620874038b2a491b50f48505c106299293fb