Spark 搜索 brings advanced full text search features to your Dataframe, Dataset and RDD. Powered by Apache Lucene.
Let's assume you have a billion records dataset you want to query on and match against another one using full text search... You do not expect an external datasource or database system than Spark, and of course with the best performances. Spark 搜索 fits your needs: it builds for all parent RDD partitions a one-2-one volatile Lucene index available during the lifecycle of your spark session across your executors local directories and RAM. Strongly typed, Spark 搜索 supports Java and Scala RDD and plans to support Python, Spark SQL and Dataset. Have a look and feel free to contribute!
- Scala
import org.apache.spark.search.rdd._ // to implicitly enhance RDD with search features
// Load some Amazon computer user reviews
val computersReviews: RDD[Review] = loadReviews("**/*/reviews_Computers.json.gz")
// Number of partition is the number of Lucene index which will be created across your cluster
.repartition(4)
// Count positive review: indexation + count matched doc with fuzzy matching
computersReviews.count("reviewText:happy OR reviewText:best OR reviewText:good OR reviewText:\"sounds great\"~")
// 搜索 for key words
computersReviews.searchList("reviewText:\"World of Warcraft\" OR reviewText:\"Civilization IV\"",
topK = 100, minScore = 10)
.foreach(println)
// /!\ Important lucene indexation is done each time a 搜索RDD is computed,
// if you do multiple operations on the same parent RDD, you might have a variable in the driver:
val computersReviews搜索RDD: 搜索RDD[Review] = computersReviewsRDD.searchRDD(
搜索Options.builder[Review]() // See all other options 搜索Options, IndexationOptions and ReaderOptions
.read((r: ReaderOptions.Builder[Review]) => r.defaultFieldName("reviewText"))
.analyzer(classOf[EnglishAnalyzer])
.build())
// Boolean queries and boosting examples returning RDD
computersReviews搜索RDD.search("(RAM OR memory) AND (CPU OR processor~)^4", 15)
.collect()
.foreach(println)
// Fuzzy matching
computersReviews.searchList("(reviewerName:Mikey~0.8) OR (reviewerName:Wiliam~0.4) OR (reviewerName:jonh~0.2)",
topKByPartition = 10)
.map(doc => s"${doc.source.reviewerName}=${doc.score}")
.foreach(println)
// RDD full text joining - example here searches for persons
// who did both computer and software reviews with fuzzy matching on reviewer name
val softwareReviews: RDD[Review] = loadReviews("**/*/reviews_Software_10.json.gz")
val matchesReviewers: RDD[(Review, Array[搜索Record[Review]])] = computersReviews.matches(
softwareReviewsRDD.filter(_.reviewerName != null).map(sr => (sr.asin, sr)),
(sr: Review) => "reviewerName:\"" + sr.reviewerName + "\"~0.4",
topK = 10)
.values
matchesReviewersRDD
.filter(_._2.nonEmpty)
.map(m => (s"Reviewer ${m._1.reviewerName} reviews computer ${m._1.asin} but also on software:",
m._2.map(h => s"${h.source.reviewerName}=${h.score}=${h.source.asin}").toList))
.collect()
.foreach(println)
// Drop duplicates
println("Dropping duplicated reviewers:")
val distinctReviewers: RDD[String] = computersReviews.searchDropDuplicates[Int, Review](
queryBuilder = queryStringBuilder(sr => "reviewerName:\"" + sr.reviewerName.replace('"', ' ') + "\"~0.4")
).map(sr => sr.reviewerName)
distinctReviewers.collect().foreach(println)
// Save then restore onto hdfs
matchesReviewersRDD.save("/tmp/hdfs-pathname")
val restored搜索RDD: 搜索RDD[Review] = 搜索RDD.load[Review](sc, "/tmp/hdfs-pathname")
// Restored index can be used as classical rdd
val topReviewer = restored搜索RDD.map(r => (r.reviewerID, 1))
.reduceByKey(_ + _)
.sortBy(_._2, ascending = false)
.take(1).head
println(s"${topReviewer._1} has submitted ${topReviewer._2} reviews")See Examples and Documentation for more details.
- Java
import org.apache.spark.search.rdd.*;
class 搜索RDDJava {
public void examples() {
System.err.println("Loading reviews...");
JavaRDD<Review> reviewsRDD = loadReviewRDD(spark, "http://snap.stanford.edu/data/amazon/productGraph/categoryFiles/reviews_Computers.json.gz");
// Create the 搜索RDD based on the JavaRDD to enjoy search features
搜索RDDJava<Review> computerReviews = 搜索RDDJava.of(reviewsRDD, Review.class);
// Count matching docs
System.err.println("Computer reviews with good recommendations: "
+ computerReviews.count("reviewText:good AND reviewText:quality"));
// List matching docs
System.err.println("Reviews with good recommendations and fuzzy: ");
搜索RecordJava<Review>[] goodReviews = computerReviews
.searchList("reviewText:recommend~0.8", 100, 0);
Arrays.stream(goodReviews).forEach(r -> System.err.println(r));
// Pass custom search options
computerReviews = 搜索RDDJava.<Review>builder()
.rdd(reviewsRDD)
.runtimeClass(Review.class)
.options(搜索Options.<Review>builder().analyzer(ShingleAnalyzerWrapper.class).build())
.build();
System.err.println("Top 100 reviews from Patosh with fuzzy with 0.5 minimum score:");
computerReviews.search("reviewerName:Patrik~0.5", 100, 0.5)
.map(搜索RecordJava::getSource)
.map(Review::getReviewerName)
.distinct()
.collect()
.foreach(r -> System.err.println(r));
System.err.println("Loading software reviews...");
JavaRDD<Review> softwareReviews = loadReviewRDD(spark, "http://snap.stanford.edu/data/amazon/productGraph/categoryFiles/reviews_Software_10.json.gz");
System.err.println("Top 10 reviews from same reviewer between computer and software:");
computerReviews.matches(softwareReviews.filter(r -> r.reviewerName != null && !r.reviewerName.isEmpty())
.mapToPair(sr -> new Tuple2<String, Review>(sr.asin, sr)),
r -> String.format("reviewerName:\"%s\"~0.4", r.reviewerName.replaceAll("[\"]", " ")), 10, 0)
.values()
.filter(matches -> matches._2.length > 0)
.map(sameReviewerMatches -> String.format("Reviewer:%s reviews computer %s and software %s (score on names matching are %s)",
sameReviewerMatches._1.reviewerName,
sameReviewerMatches._1.asin,
Arrays.stream(sameReviewerMatches._2).map(h -> h.source.asin).collect(toList()),
Arrays.stream(sameReviewerMatches._2).map(h -> h.source.reviewerName + ":" + h.score).collect(toList())
))
.collect()
.foreach(matches -> System.err.println(matches));
// Save and search reload example
搜索RDDJava.of(softwareReviews.repartition(8), Review.class)
.save("/tmp/hdfs-pathname");
搜索RDDJava<Review> restored搜索RDD = 搜索RDDJava
.load(sc, "/tmp/hdfs-pathname", Review.class);
System.err.println("Software reviews with good recommendations: "
+ restored搜索RDD.count("reviewText:good AND reviewText:quality"));
}
}See Examples and Documentationfor more details.
- Python (In progress)
from pyspark import SparkContext
import pysparksearch
data = [{"firstName": "Geoorge", "lastName": "Michael"},
{"firstName": "Bob", "lastName": "Marley"},
{"firstName": "Agnès", "lastName": "Bartoll"}]
sc = SparkContext()
sc.parallelize(data).count("firstName:agnes~")- Scala
import org.apache.spark.search.sql._
val sentences = spark.read.csv("...")
sentences.count("sentence:happy OR sentence:best OR sentence:good")
// coming soon: 搜索SparkStrategy/LogicPlan & column enhanced with search
sentences.where($"sentence".matches($"searchKeyword" ))All benchmarks run under AWS EMR with 3 Spark workers EC2 m5.xlarge and/or 3 r5.large.elasticsearch data nodes for AWS Elasticsearch. The general use cases is to match company names against two data sets (7M vs 600K rows)
| Feature | 搜索RDD | Elasticsearch Hadoop | LuceneRDD | Spark regex matches (no score) |
|---|---|---|---|---|
| Index + Count matches | 51s | 486s (*) | 400s | 12s |
| Index + Entity matching | 128s | 719s (*) | 597s | NA (>1h) |
DISCLAIMER Benchmarks methodology or related results may improve, feel free to submit a pull request.
(*) Results of elasticsearch hadoop benchmark must be carefully reviewed, contribution welcomed
- 搜索RDD#searchJoin renamed to 搜索RDD#matches as it does automatically the reduction in addition of a simple join.
- Fix matches was using only one core & improve join and dropDuplicate performances drastically
- Scala 2.12 by default
- Fix deployment descriptor for scala 2.11
- 搜索RDD is now iterable as a classical RDD, reloaded RDD can now be used as any other RDD
- Upgrade support matrix from spark-2.4.8 & hadoop-2.10.1 to spark-3.1.2 & hadoop-3.3.1, built by default for scala 2.12
- Enable caching of search index rdd only for yarn cluster, and as an option.
- Remove scala binary version in parent module artifact name
- Expose 搜索RDD as a public API to ease Dataset binding and hdfs reloading
- Fix and enhance 搜索 Java RDD API
- Fix string query builder does not support analyzer
- Switch to multi modules build: core, sql, examples, benchmark
- Improve the github build with running examples against a spark cluster in docker
- Improve licence header checking
- RDD lineage works the same on all DAG Scheduler (Yarn/Standalone): 搜索IndexRDD computes zipped index per partition for the next rdd
- CI tests examples under Yarn and Standalone cluster mode
- Fix default field where not used under certain circumstances
- Fix 搜索RDD#searchDropDuplicate method
- Save/Restore search RDD to/from HDF
- Yarn support and tested over AWS EMR
- Adding and running benchmark examples with alternatives libraries on AWS EMR
- Support of spark 3.0.0
- Optimize searchJoin for small num partition
- Fix searchJoin on multiple partitions
- Released to maven central
- First stable version of the Scala Spark 搜索 RDD
- Support of
搜索RDD#searchJoin(RDD, S => String)- join 2 RDD by matching queries - Support of
搜索RDD#dropDuplicates(S => String)- deduplicate an RDD based on matching query
- Support of
搜索RDD#count(String)- count matching hits - Support of
搜索RDD#searchList(String)- search matching records as list - Support of
搜索RDD#search(String)- search matching records as RDD
- Maven
<dependency>
<groupId>io.github.phymbert</groupId>
<artifactId>spark-search_2.12</artifactId>
<version>${spark.search.version}</version>
</dependency>- Gradle
implementation 'io.github.phymbert:spark-search_2.12:$spark搜索Version'git clone https://github.com/phymbert/spark-search.git
cd spark-search
mvn clean verify