Windows单机安装配置mongodb+hadoop+spark+pyspark用于大数据分析

news2024/11/27 2:39:35

版本选择

根据Spark Connector：org.mongodb.spark:mongo-spark-connector_2.13:10.3.0 的前提要求
在这里插入图片描述
这里选择使用最新的MongoDB 7.0.12社区版
https://www.mongodb.com/try/download/community

Spark使用最新的3.5.2
https://dlcdn.apache.org/spark/spark-3.5.2/spark-3.5.2-bin-hadoop3.tgz
官网下载比较慢，可以从阿里镜像源下载：
https://mirrors.aliyun.com/apache/spark/spark-3.5.2/?spm=a2c6h.25603864.0.0.52d721049dSJJZ
在这里插入图片描述

Java使用Java8最新更新
https://www.oracle.com/cn/java/technologies/javase/javase8u211-later-archive-downloads.html

PySpark 一般会与 Hadoop 环境一起运行 , 如果在 Windows 中没有安装 Hadoop 运行环境 , 就会报错误 ;
Hadoop 发布版本在https://hadoop.apache.org/releases.html 页面可下载 ;
官网下载比较慢，可以从阿里镜像源下载：
https://mirrors.aliyun.com/apache/hadoop/common/hadoop-3.3.6/

winutils.exe是一个Windows平台上的实用工具，它是Apache Hadoop项目的一部分。Apache Hadoop是一个开源的分布式计算框架，用于处理大规模数据集的分布式存储和处理。winutils.exe主要用于在Windows环境下运行Hadoop相关的任务和操作。

winutils可以从如下github仓库下载：
https://github.com/cdarlint/winutils
在这里插入图片描述
由于winutils最新只支持hadoop-3.3.6，所以上面Hadoop下载的也是这个版本而不是最新版；

都下载后如下图
在这里插入图片描述

安装

mongodb和jdk直接按默认选项安装即可。

配置

Java环境配置

参考文章：Windows如何安装JDK
来自 https://blog.csdn.net/lcl17779740668/article/details/137992141?spm=1001.2014.3001.5502

Win+R键打开运行窗口，输入cmd，命令行串口输入java -version验证是否安装成功。

Hadoop配置

Hadoop下载后，解压即可，然后记录解压后的路径并配置环境变量
系统变量创建HADOOP_HOME变量值：hadoop安装的路径
系统变量中的Path添加：%HADOOP_HOME%\bin

将上面下载的winutils中的hadoop-3.3.6/bin文件夹下的所有文件复制到对应%HADOOP_HOME%\bin文件夹中并替换原有文件。

Win+R键打开运行窗口，输入cmd，命令行串口输入hadoop -version验证是否安装成功。

可能遇到的报错：“Error JAVA_HOME is incorrectly set.”
根因：JAVA_HOME的值有空格
解决方式一：修改系统环境变量C:\Program Files\Java\jdk-1.8为：C:\Progra~1\Java\jdk-1.8
解决方式二：直接更改hadoop对应etc\hadoop\hadoop-env.cmd 脚本中的 JAVA_HOME 为C:\Progra~1\Java\jdk-1.8
在这里插入图片描述
参考：DOS命令空格问题解决办法
来自 https://blog.csdn.net/youdaodao/article/details/89473558

Spark配置

Spark下载后，解压即可，然后记录解压后的路径并配置环境变量
新建系统变量》变量名：SPARK_HOME 变量值：spark安装的路径

系统变量path中新建两个变量值 %SPARK_HOME%\bin %SPARK_HOME%\sbin

Win+R键打开运行窗口，输入cmd，命令行串口输入spark-shell检查spark是否安装成功

安装pyspark

python环境使用的是python 3.8.10
pip install pyspark
在这里插入图片描述

使用Jupyter Notebook进行Spark+ MongoDB测试

from pyspark.sql import SparkSession

my_spark = SparkSession \
    .builder \
    .appName("myApp") \
    .config("spark.mongodb.read.connection.uri", "mongodb://localhost:27017/local.startup_log") \
    .config("spark.mongodb.write.connection.uri", "mongodb://localhost:27017/local.FSHeight") \
    .config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.13:10.3.0") \
    .getOrCreate()
dataFrame = my_spark.read.format("mongodb").load()
dataFrame.printSchema()

从local.startup_log读取数据，printSchema输出正常。

dataFrame.show()
dataFrame.count()

以上两个方法均报错：

Py4JJavaError: An error occurred while calling o42.showString.
: java.lang.NoSuchMethodError: org.apache.spark.sql.catalyst.encoders.ExpressionEncoder.resolveAndBind(Lscala/collection/immutable/Seq;Lorg/apache/spark/sql/catalyst/analysis/Analyzer;)Lorg/apache/spark/sql/catalyst/encoders/ExpressionEncoder;
	at com.mongodb.spark.sql.connector.schema.SchemaToExpressionEncoderFunction.apply(SchemaToExpressionEncoderFunction.java:97)
	at com.mongodb.spark.sql.connector.schema.RowToInternalRowFunction.<init>(RowToInternalRowFunction.java:41)
	at com.mongodb.spark.sql.connector.schema.BsonDocumentToRowConverter.<init>(BsonDocumentToRowConverter.java:100)
	at com.mongodb.spark.sql.connector.read.MongoBatch.<init>(MongoBatch.java:47)
	at com.mongodb.spark.sql.connector.read.MongoScan.toBatch(MongoScan.java:79)
	at org.apache.spark.sql.execution.datasources.v2.BatchScanExec.batch$lzycompute(BatchScanExec.scala:45)
	at org.apache.spark.sql.execution.datasources.v2.BatchScanExec.batch(BatchScanExec.scala:45)
	at org.apache.spark.sql.execution.datasources.v2.BatchScanExec.inputPartitions$lzycompute(BatchScanExec.scala:59)
	at org.apache.spark.sql.execution.datasources.v2.BatchScanExec.inputPartitions(BatchScanExec.scala:59)
	at org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanExecBase.supportsColumnar(DataSourceV2ScanExecBase.scala:179)
	at org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanExecBase.supportsColumnar$(DataSourceV2ScanExecBase.scala:175)
	at org.apache.spark.sql.execution.datasources.v2.BatchScanExec.supportsColumnar(BatchScanExec.scala:36)
	at org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy.apply(DataSourceV2Strategy.scala:147)
	at org.apache.spark.sql.catalyst.planning.QueryPlanner.$anonfun$plan$1(QueryPlanner.scala:63)
	at scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:486)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:492)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:491)

报错对应的是如下位置
在这里插入图片描述
从spark-3.5.2-bin-hadoop3\jars路径下找到了spark-sql_2.12-3.5.2.jar。
解压缩后发现并没有对应的org.apache.spark.sql.catalyst.encoders.ExpressionEncoder.resolveAndBind

根因就在这里

因为下载的spark编译包中spark-3.5.2-bin-hadoop3\jars都是基于scala 2.12的，
将org.mongodb.spark:mongo-spark-connector_2.13:10.3.0换成org.mongodb.spark:mongo-spark-connector_2.12:10.3.0问题解决

from pyspark.sql import SparkSession

my_spark = SparkSession \
    .builder \
    .appName("myApp") \
    .config("spark.mongodb.read.connection.uri", "mongodb://localhost:27017/local.startup_log") \
    .config("spark.mongodb.write.connection.uri", "mongodb://localhost:27017/local.FSHeight") \
    .config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.13:10.3.0") \
    .getOrCreate()
# 从上述spark.mongodb.read.connection.uri读取数据    
df = my_spark.read.format("mongodb").load()
# 将读取的数据写入上述spark.mongodb.write.connection.uri，mode可选overwrite or append
df.write.format("mongodb").mode("overwrite").save()
# 重新从数据库读取数据，并查看
dataFrame = my_spark.read\
                 .format("mongodb")\
                 .option("database", "local")\
                 .option("collection", "FSHeight")\
                 .load()
dataFrame.printSchema()
dataFrame.count()
dataFrame.show()