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