Spark——Spark读写MongoDB

文章目录

Spark直连MongoDB
- 1. 通过SparkConf
- 2. 通过ReadConfig
- 3. 通过DataFrameReader
Hive中创建映射表
mongoexport导出文件
参考

如果我们要想通过Spark来处理MongoDB中数据，可以采用以下3种方式：

Spark直连MongoDB
Hive中创建映射表
使用MongoDB Export导出文件

Spark直连MongoDB

Spark直连MongoDB有以下3种方式，所需要的Maven依赖：

<dependency>
    <groupId></groupId>
    <artifactId>mongo-spark-connector_2.11</artifactId>
    <version>2.4.0</version>
</dependency>

<dependency>
    <groupId></groupId>
    <artifactId>mongo-java-driver</artifactId>
    <version>3.12.7</version>
</dependency>

1. 通过SparkConf

MongoDB相关配置在SparkSession初始化时配置，实现代码：

val spark = SparkSession
	.builder()
	.appName("Spark-MongoDB")
	.config("", "mongodb://username:password@host/db_name.collection_name?authSource=admin")
	.config("", "MongoSamplePartitioner")
	.enableHiveSupport()
	.getOrCreate()
	
val rdd: MongoRDD[Document = MongoSpark.load(spark.sparkContext)

spark.close()

2. 通过ReadConfig

MongoDB相关配置通过类进行配置，实现代码：

val spark = SparkSession
	.builder()
	.appName("Spark-MongoDB")
	.enableHiveSupport()
	.getOrCreate()

import com.mongodb.spark.config._
val readConfig = ReadConfig(
	Map(
		"uri" -> "mongodb://username:password@host/",
		"database" -> "db_name",
		"collection" -> "collection_name"
	)
)
val rdd: MongoRDD[Document] = MongoSpark.load(spark.sparkContext, readConfig)

spark.close()

3. 通过DataFrameReader

MongoDB相关配置通过Spark DataFrameReader进行配置，实现代码：

val spark = SparkSession
	.builder()
	.appName("Spark-MongoDB")
	.enableHiveSupport()
	.getOrCreate()
	
val df: DataFrame = spark.read
		.format("")
		.option("uri", "mongodb://username:password@host/db_name.collection_name?authSource=admin")
		.load()

spark.close()

Hive中创建映射表

这种方式是先通过在Hive中创建MongoDB中要读取的collection所对应的映射表，然后用Spark读取这个映射表就可以获取MongoDB中的数据了。这种创建映射表的方式，数据还是存储在MongoDB中的，Hive中没有任何原始数据，只有对应的映射表。

第一步：创建映射表需要如下依赖包，需要将jar包复制到Hive安装目录的lib文件夹下

mongo-hadoop-core-2.0.
mongo-hadoop-hive-2.0.
mongo-java-driver-3.8.

第二步：在Hive中创建映射表

create table rds.sdhh_contacts_mongo( 
  user_id string,
  user_name string,
  updated_at timestamp,
  created_at timestamp
) 
stored by ''
with serdeproperties(''='{"user_id":"user_name","created_at":"created_at"}')
tblproperties(''='mongodb://username:password@host:port/?authsource=admin');

mongoexport导出文件

mongoexport是MongoDB的一个命令行工具，可以将MongoDB中的collection数据导出为JOSN或CSV文件。然后再将这些文件上传到HDFS上，就可以用Spark进行解析了。

具体步骤如下：

MongoDB安装目录下执行mongoexport命令，导出collection到本地
```
./mongoexport --host host_name  --port port_no -u user_name -p password -d db_name -c collection_anme -f col1,col2 -q '{"startupTime":{$gte:ISODate("2020-09-17T16:00:00.000Z"),$lte:ISODate("2020-09-18T15:59:59.999Z")}}'  -o /opt/export.json
```
- –host：MongoDB安装的主机名
- –port：MongoDB进程端口号
- -u：用户名
- -p：密码
- -d：数据库名
- -c：collection名
- -f：指定要查询的列
- -q：指定查询条件
- -o：指定输出文件
上传导出的json文件到HDFS
```
hadoop fs -put /opt/export.json /
```

Spark读取HDFS文件并处理

val df: DataFrame = spark.read.json("/")

参考

/spark-connector/v2.4/scala/read-from-mongodb
/spark-connector/v2.4/configuration#std-label-spark-output-conf
/mongodb/mongo-spark
/database-tools/mongoexport/#

秒客网