Fork me on GitHub

Spark教程(二)Spark连接MongoDB

1
这是崔斯特的第六十五篇原创文章

学习Spark (๑• . •๑)

如何导入数据

数据可能有各种格式,虽然常见的是HDFS,但是因为在Python爬虫中数据库用的比较多的是MongoDB,所以这里会重点说说如何用spark导入MongoDB中的数据。

当然,首先你需要在自己电脑上安装spark环境,简单说下,在这里下载spark,同时需要配置好JAVAScala环境。

这里建议使用Jupyter notebook,会比较方便,在环境变量中这样设置

1
2
PYSPARK_DRIVER_PYTHON=jupyter
PYSPARK_DRIVER_PYTHON_OPTS=notebook ./bin/pyspark

如果你的环境中有多个Python版本,同样可以制定你想要使用的解释器,我这里是python36,根据需求修改。

1
PYSPARK_PYTHON=/usr/bin/python36

启动命令

进入spark根目录,./bin/pyspark这是最简单的启动命令,默认会打开Python的交互式解释器,但是由于我们上面有设置过,会打开Jupyter notebook,接下来变成会方便很多。

先来看看最简单的例子:

1
2
3
4
5
6
7
8
9
10
11
12
13
>>> textFile = spark.read.text("README.md")
>>> textFile.count() # Number of rows in this DataFrame
126
>>> textFile.first() # First row in this DataFrame
Row(value=u'# Apache Spark')
>>> linesWithSpark = textFile.filter(textFile.value.contains("Spark"))
>>> textFile.filter(textFile.value.contains("Spark")).count() # How many lines contain "Spark"?
15

这里有我之前写过的例子,可以照着写一遍 basic_exercise

我们的启动方式是./bin/pyspark,我们可以家后面加很多参数,比如说如若我们要连接MongoDB,就需要这样

完整的可以参考Spark Connector Python Guide

1
2
3
./bin/pyspark --conf "spark.mongodb.input.uri=mongodb://127.0.0.1/test.myCollection?readPreference=primaryPreferred" \
--conf "spark.mongodb.output.uri=mongodb://127.0.0.1/test.myCollection" \
--packages org.mongodb.spark:mongo-spark-connector_2.11:2.3.0

这里有两个uri,分别是inputoutput,对应读取的数据库和写入的数据库,最后面的packages相当于引入的包的名字,我一般喜欢在代码中定义。

读取/保存数据

这里我们可以增加参数option,在这里设置想要读取的数据库地址,注意格式。

读取数据

1
2
df = spark.read.format("com.mongodb.spark.sql.DefaultSource").option("uri",
"mongodb://127.0.0.1/people.contacts").load()

保存数据

1
2
3
people.write.format("com.mongodb.spark.sql.DefaultSource").mode("append").option("uri",
"mongodb://127.0.0.1/people.contacts").option("database",
"people").option("collection", "contacts").save()

简单对比下,option还可以定义databasecollection,这样就不需要在启动Spark时定义。

以上是官网推荐的连接方式,这里需要说的是另一种,如果我没有从命令行中启动,而是直接新建一个py文件,该如何操作?

搜索相关资料后,发现是这样

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = 'zhangslob'
import os
from pyspark.sql import SparkSession
# set PYSPARK_PYTHON to python36
os.environ['PYSPARK_PYTHON'] = '/usr/bin/python36'
# load mongo data
input_uri = "mongodb://127.0.0.1:spark.spark_test"
output_uri = "mongodb://127.0.0.1:spark.spark_test"
my_spark = SparkSession\
.builder\
.appName("MyApp")\
.config("spark.mongodb.input.uri", input_uri)\
.config("spark.mongodb.output.uri", output_uri)\
.config('spark.jars.packages','org.mongodb.spark:mongo-spark-connector_2.11:2.2.0')\
.getOrCreate()
df = my_spark.read.format('com.mongodb.spark.sql.DefaultSource').load()

必须要增加默认设置('spark.jars.packages','org.mongodb.spark:mongo-spark-connector_2.11:2.2.0'),否则会报错。

本文标题:Spark教程(二)Spark连接MongoDB

文章作者:小歪

发布时间:2018年09月03日 - 22:09

最后更新:2018年09月03日 - 22:09

原始链接:https://zhangslob.github.io/2018/09/03/Spark教程(二)Spark连接MongoDB/

许可协议: 署名-非商业性使用-禁止演绎 4.0 国际 转载请保留原文链接及作者。

小歪 wechat
欢迎关注公众号