如何过滤 spark 链接 mongodb 时 加载数据量 20C

目前使用 的是mongodb官方提供的一个链接spark的jar包,代码如下
SparkConf conf = new SparkConf().set(...)//设置初始化属性
JavaSparkContext jsc = new JavaSparkContext(conf);
//下面这步加载数据
Dataset explicitDF = MongoSpark.load(jsc).toDF();
由于是这样直接加载的,根本 没地方在加载之前写过滤条件来控制加载的数据量的,所以导致直接加载了整表的数据,十分缓慢,各位有 什么方法或者建议提供参考下么,谢谢啦

2个回答

val spark = SparkSession.builder

.appName(this.getClass.getName().stripSuffix("$"))

.getOrCreate()

val inputUri="mongodb://test:pwd123456@192.168.0.1:27017/test.articles"

val df = spark.read.format("com.mongodb.spark.sql").options(

Map("spark.mongodb.input.uri" -> inputUri,

"spark.mongodb.input.partitioner" -> "MongoPaginateBySizePartitioner",

"spark.mongodb.input.partitionerOptions.partitionKey" -> "_id",

"spark.mongodb.input.partitionerOptions.partitionSizeMB"-> "32"))

.load()

val currentTimestamp = System.currentTimeMillis()

val originDf = df.filter(df("updateTime") < currentTimestamp && df("updateTime") >= currentTimestamp - 1440 * 60 * 1000)

.select("_id", "content", "imgTotalCount").toDF("id", "content", "imgnum")

可以使用mongoexport导出需要的信息到一个文件中,然后再用spark解析文件,速度也要快不少。

Csdn user default icon
上传中...
上传图片
插入图片
抄袭、复制答案,以达到刷声望分或其他目的的行为,在CSDN问答是严格禁止的,一经发现立刻封号。是时候展现真正的技术了!
立即提问