pyspark操作MongoDB的方法步骤
作者:小歪的博客 时间:2021-05-02 11:12:12
如何导入数据
数据可能有各种格式,虽然常见的是HDFS,但是因为在Python爬虫中数据库用的比较多的是MongoDB,所以这里会重点说说如何用spark导入MongoDB中的数据。
当然,首先你需要在自己电脑上安装spark环境,简单说下,在这里下载spark,同时需要配置好JAVA,Scala环境。
这里建议使用Jupyter notebook,会比较方便,在环境变量中这样设置
PYSPARK_DRIVER_PYTHON=jupyter PYSPARK_DRIVER_PYTHON_OPTS=notebook ./bin/pyspark
如果你的环境中有多个Python版本,同样可以制定你想要使用的解释器,我这里是python36,根据需求修改。
PYSPARK_PYTHON=/usr/bin/python36
pyspark对mongo数据库的基本操作 (๑• . •๑)
有几点需要注意的:
不要安装最新的pyspark版本,请安装
pip3 install pyspark==2.3.2
spark-connector
与平常的MongoDB写法不同,格式是:mongodb://127.0.0.1:database.collection
如果计算数据量比较大,你的电脑可能会比较卡,^_^
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@author: zhangslob
@file: spark_count.py
@time: 2019/01/03
@desc:
不要安装最新的pyspark版本
`pip3 install pyspark==2.3.2`
更多pyspark操作MongoDB请看https://docs.mongodb.com/spark-connector/master/python-api/
"""
import os
from pyspark.sql import SparkSession
# set PYSPARK_PYTHON to python36
os.environ['PYSPARK_PYTHON'] = '/usr/bin/python36'
# load mongodb data
# 格式是:"mongodb://127.0.0.1:database.collection"
input_uri = "mongodb://127.0.0.1:27017/spark.spark_test"
output_uri = "mongodb://127.0.0.1:27017/spark.spark_test"
# 创建spark,默认使用本地环境,或者"spark://master:7077"
spark = SparkSession \
.builder \
.master("local") \
.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()
def except_id(collection_1, collection_2, output_collection, pipeline):
"""
计算表1与表2中不同的数据
:param collection_1: 导入表1
:param collection_2: 导入表2
:param output_collection: 保存的表
:param pipeline: MongoDB查询语句 str
:return:
"""
# 可以在这里指定想要导入的数据库,将会覆盖上面配置中的input_uri。下面保存数据也一样
# .option("collection", "mongodb://127.0.0.1:27017/spark.spark_test")
# .option("database", "people").option("collection", "contacts")
df_1 = spark.read.format('com.mongodb.spark.sql.DefaultSource').option("collection", collection_1) \
.option("pipeline", pipeline).load()
df_2 = spark.read.format('com.mongodb.spark.sql.DefaultSource').option("collection", collection_2) \
.option("pipeline", pipeline).load()
# df_1有但是不在 df_2,同理可以计算df_2有,df_1没有
df = df_1.subtract(df_2)
df.show()
# mode 参数可选范围
# * `append`: Append contents of this :class:`DataFrame` to existing data.
# * `overwrite`: Overwrite existing data.
# * `error` or `errorifexists`: Throw an exception if data already exists.
# * `ignore`: Silently ignore this operation if data already exists.
df.write.format("com.mongodb.spark.sql.DefaultSource").option("collection", output_collection).mode("append").save()
spark.stop()
if __name__ == '__main__':
# mongodb query, MongoDB查询语句,可以减少导入数据量
pipeline = "[{'$project': {'uid': 1, '_id': 0}}]"
collection_1 = "spark_1"
collection_2 = "spark_2"
output_collection = 'diff_uid'
except_id(collection_1, collection_2, output_collection, pipeline)
print('success')
完整代码地址: spark_count_diff_uid.py
来源:https://zhangslob.github.io/2019/01/03/pyspark操作MongoDB/
标签:pyspark,操作,MongoDB


猜你喜欢
Python 学习教程之networkx
2023-10-11 01:49:07
对python字典过滤条件的实例详解
2022-01-28 14:02:30
利用python进行数据加载
2022-05-11 19:55:00
centos yum php 7.x 无需删除升级的方法
2023-11-20 14:48:08

python脚本框架webpy的url映射详解
2021-10-27 18:55:05
Python 远程开关机的方法
2022-06-19 18:12:30

python中MethodType方法介绍与使用示例
2022-09-08 03:28:50
用python批量移动文件
2022-12-21 10:48:43
一文搞懂Python中列表List和元组Tuple的使用
2022-10-09 02:24:58

简单的PHP图片上传程序
2024-05-22 10:06:54
Django分页器的用法你都了解吗
2022-06-06 21:18:39
asp的分词技术
2007-08-25 17:50:00
SQL Server 2008中的新日期数据类型
2009-03-16 15:05:00

Go Comparable Type原理深入解析
2024-02-21 03:47:16
python cv2图像质量压缩的算法示例
2023-08-16 19:09:02

MySQL 压缩的使用场景和解决方案
2024-01-19 15:45:58

支持在线写SQL的Oracle学习免费网站推荐!(个人常使用)
2023-07-24 09:06:51

Oracle 数据库自动存储管理-安装配置
2009-05-24 19:15:00
localResizeIMG先压缩后使用ajax无刷新上传(移动端)
2024-05-09 10:35:26
Python数据分析之双色球统计两个红和蓝球哪组合比例高的方法
2021-05-14 01:15:57
