如何用Apache Spark高效加载大型Amazon商品评论数据集到MongoDB

针对126GB的亚马逊行分隔JSON数据集，结合Schema和MongoDB结构优化加载流程，核心优化点如下：

1. 显式定义Schema，消除自动推断开销 #

Spark自动推断Schema需要扫描部分数据，对于大文件会产生额外耗时。提前定义匹配数据集的Schema，直接用Spark内置JSON Reader加载，跳过推断步骤：

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, FloatType, LongType, ArrayType

# 匹配亚马逊评论数据集的完整Schema（根据实际字段调整）
amazon_schema = StructType([
    StructField("reviewerID", StringType(), nullable=True),
    StructField("asin", StringType(), nullable=True),
    StructField("reviewerName", StringType(), nullable=True),
    StructField("helpful", ArrayType(IntegerType()), nullable=True),
    StructField("reviewText", StringType(), nullable=True),
    StructField("overall", FloatType(), nullable=True),
    StructField("summary", StringType(), nullable=True),
    StructField("unixReviewTime", LongType(), nullable=True),
    StructField("reviewTime", StringType(), nullable=True)
])

# 用Spark内置JSON Reader加载，指定行分隔符和Schema
reviews_df = spark.read.schema(amazon_schema)\
    .option("lineSep", "\n")\
    .json("X:/reviews.json.gz")

说明：Spark内置JSON Reader比手动RDD.map(json.loads)更高效，支持并行解析，减少序列化开销

2. 调整读取并行度，利用集群资源 #

单一大压缩文件的默认分区数可能不足，通过以下方式提升并行处理能力：

• 在SparkSession初始化时调整分区大小：

.config("spark.sql.files.maxPartitionBytes", "128m")  # 按集群核心数调整，建议设置为128-256MB/分区

• 根据集群核心数重新分区：

# 假设集群有20个核心，设置分区数为核心数的2-3倍
reviews_df = reviews_df.repartition(60)

3. 优化MongoDB写入配置 #

针对MongoDB的写入特性，调整连接器参数减少IO开销：

• 增大批量写入大小，降低请求频次：

.config("spark.mongodb.output.batchSize", "10000")  # 默认1000，可根据MongoDB性能调整

• 对齐Spark分区与MongoDB分片键：
  如果MongoDB集合按asin或reviewerID分片，让Spark DataFrame预先按该字段分区，避免跨分片写入：

reviews_df = reviews_df.repartition("asin")

• 调整写入一致性级别（可选）：
  若不需要强一致性，设置写入关注级别为1，提升写入速度：

.config("spark.mongodb.output.writeConcern.w", "1")

4. 移除冗余数据转换 #

原代码中RDD.map(json.loads).toDF()的转换是冗余的，直接用Spark DataFrame API加载数据，避免RDD与DataFrame之间的序列化/反序列化开销。

5. 配置合理的Spark资源 #

根据集群硬件调整资源参数，避免内存不足或资源浪费：

spark = SparkSession.builder \
    .appName("AmazonReviewsToMongoDB") \
    .config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.12:3.0.1") \
    .config("spark.executor.instances", "10")  # 执行器数量
    .config("spark.executor.cores", "4")  # 每个执行器核心数
    .config("spark.executor.memory", "16g")  # 每个执行器内存
    .config("spark.driver.memory", "8g")  # 驱动内存
    # 其他MongoDB和Schema配置
    .getOrCreate()

6. 预测试验证 #

正式运行前用小数据集验证Schema匹配和写入逻辑：

test_df = spark.read.schema(amazon_schema)\
    .option("lineSep", "\n")\
    .json("X:/reviews.json.gz")\
    .limit(1000)
test_df.write.format("mongo").mode("overwrite").save()

内容的提问来源于stack exchange，提问作者Hammad Javaid