E-Commerce Search: Implementing Store Rating Weights

We enhanced the search system with a store rating field and built a data synchronization and ranking weighting mechanism. The approach consists of three parts:

Data source: Store ratings are computed offline by the platform's data team and stored in a dedicated store score table (e.g., daily-updated ratings database). Ratings range from 0 to 5 with one decimal place of precision. To keep search updated, we use Kafka message queue to push rating changes to the search system in near-real-time. We also retain batch processing to periodically refresh all data and ensure consistency.
Index extension: We add a store_score field to product search indices to hold the rating of each product's store. The field is populated from the store score table when building new indices; for existing indexed products, we use batch or incremental updates to backfill values.
Ranking weighting: During search queries, the engine first calculates a base score from text relevance and product attribute matching. In the secondary ranking phase, we apply store rating weights: if a product's store has a rating below a threshold (e.g., 2.5), we multiply its base score by a penalty coefficient (e.g., 0.95, a 5% reduction); otherwise the coefficient is 1. Thresholds and penalty coefficients are all configurable for later tuning.

The following sections detail each component with Python code examples.

Kafka Message Consumption

To synchronize store rating data in real-time, we built a Kafka consumer to subscribe to rating update topics. After the data team computes new store ratings, they send changes to a Kafka topic (e.g., store-rating-topic). The search system's consumption module subscribes to that topic, continuously listening for new messages and processing updates.

Here is the Kafka consumer code:

from kafka import KafkaConsumer

def consume_shop_ratings():
    consumer = KafkaConsumer(
        'shop-rating-updates',
        bootstrap_servers=['kafka1:9092', 'kafka2:9092'],
        auto_offset_reset='latest',
        group_id='shop-rating-group',
        enable_auto_commit=True,
        consumer_timeout_ms=1000
    )

    for message in consumer:
        shop_id, rating = message.value.decode('utf-8').split(",")
        rating = float(rating)
        update_shop_rating(shop_id, rating)

def update_shop_rating(shop_id, rating):
    # 示例更新缓存或数据库，这里用简单打印代替
    print(f"Shop {shop_id} rating updated to {rating}")

if __name__ == "__main__":
    consume_shop_ratings()

The consumer subscribes to store-rating-topic and continuously polls for store rating updates. Upon receiving a message, it parses the store ID and new rating value, then calls the updateStoreScore method. We can hold the updated rating temporarily in an in-memory cache or queue, then asynchronously trigger updates to related products in the search index (e.g., calling an index update interface to update the store_score field for those products).

In production, the consumer must handle idempotency (repeated messages must not cause data inconsistency), manage exceptions (Kafka connection failures, retries), and batch updates for high-frequency stores. The code above is simplified; it omits complete error handling and batch control logic.

Batch Store Rating Synchronization

Beyond real-time messaging, we also implemented batch processing to periodically sync the full store rating dataset. This is useful when initially launching the feature (backfilling store_score for all products) or as a fallback mechanism to correct missed updates. Batch processing reads directly from the store score table (e.g., in MySQL or Hive) to obtain all store ratings, then updates the search index.

Batch tasks typically run in the early morning hours and perform these steps:

Read all current store ratings from the store score database table.
Load store IDs and ratings into memory (e.g., in a Map) or generate an update instruction list.
Call the index update module to update each store's related products in the index to the latest rating.

Here is sample code for batch reading store ratings:

import pymysql

def fetch_shop_ratings():
    connection = pymysql.connect(
        host='your-db-host',
        user='your-username',
        password='your-password',
        database='shop_rating_db'
    )

    shop_rating_map = {}
    try:
        with connection.cursor() as cursor:
            cursor.execute("SELECT shop_id, rating FROM daily_shop_ratings")
            for shop_id, rating in cursor.fetchall():
                shop_rating_map[shop_id] = rating
    finally:
        connection.close()

    return shop_rating_map

if __name__ == "__main__":
    ratings = fetch_shop_ratings()
    print(f"Total shops loaded: {len(ratings)}")
    # 后续调用索引更新函数

This batch code connects via pymysql to the database, executes a SQL query to fetch all store ratings, and stores them in a map. Once we have the full dataset, we can iterate through it to update the search index for each store's related products (the index update section below covers this in detail). Batch processing must attend to performance and resource usage; consider paginated or streaming reads to avoid loading too much data into memory at once. Also ensure that bulk index updates have minimal impact on search service availability—execute during low-traffic periods and control commit frequency appropriately.

Index Field Extension and Update

With store rating data in hand, we need to write it into the search index for use during ranking. First, we add a new field to the index schema, such as store_score, using a numeric type (e.g., float) to store rating values. For products already in the index, we must perform an update to populate this field. We use two strategies: full rebuild or incremental update. Full rebuild re-indexes all product data with store ratings included (recommended initially or when data volume is manageable). Incremental update modifies a subset of the existing index, leveraging Kafka messages or batch results to update products from changed stores one at a time.

To demonstrate index updates, here is example code using pysolr for atomic field updates. pysolr supports updating specific fields of existing documents without reconstructing the entire document. Assuming we have identified stores needing updates, their new ratings, and product ID lists for each store, we can update the Solr index as follows:

import pysolr

solr_url = 'http://your-solr-host:8983/solr/product_core'
solr_client = pysolr.Solr(solr_url, always_commit=True)

def update_shop_rating_index(shop_id, rating, product_ids):
    docs = []
    for pid in product_ids:
        doc = {
            "id": pid,
            "shop_rating": {"set": rating}
        }
        docs.append(doc)
    solr_client.add(docs)
    print(f"Updated shop {shop_id} products with rating {rating}")

if __name__ == "__main__":
    example_shop_id = 'shop789'
    example_rating = 4.2
    example_products = ['prod001', 'prod002', 'prod003']
    update_shop_rating_index(example_shop_id, example_rating, example_products)

This code connects to the Solr server via pysolr, then constructs dictionaries to perform field updates.

For other search engines like Elasticsearch, the implementation differs in details but follows the same logic: ensure the index has a store_score field, then bulk update document values for that field. Regardless of search engine choice, the goal is identical: each product document must carry its store rating for downstream ranking calculations.

Search Ranking Weighting

Once the index contains store rating fields, we can apply our weighting policy during queries. In the secondary ranking phase, we retrieve each result product's base score and store rating, then apply a threshold-based decision on whether to apply a penalty. The base score typically combines multiple relevance factors—keyword match score, brand preference bonuses, category match bonuses—summed into an initial score. We then apply store rating weights: if the store rating is below the threshold, multiply by a penalty coefficient; otherwise, keep the coefficient at 1.

For example, we set the threshold to 2.5 points and penalty coefficient to 0.95 (5% score reduction). If a product's store rates 2.0 (below 2.5), the final score is base_score × 0.95. If the store rates 3.5 (above threshold), the final score is base_score × 1.0 (unchanged). This way, products from low-rating stores receive slightly lower scores and rank slightly lower overall.

Here is simplified code for secondary ranking weighting:

THRESHOLD_RATING = 2.5
PENALTY_FACTOR = 0.95

def calculate_final_score(base_score, shop_rating):
    if shop_rating < THRESHOLD_RATING:
        return base_score * PENALTY_FACTOR
    return base_score

if __name__ == "__main__":
    products = [
        {"id": "prod001", "base_score": 80.0, "shop_rating": 2.0},
        {"id": "prod002", "base_score": 75.0, "shop_rating": 3.0},
        {"id": "prod003", "base_score": 90.0, "shop_rating": 4.5},
    ]

    for product in products:
        final_score = calculate_final_score(product["base_score"], product["shop_rating"])
        print(f"Product {product['id']} final score: {final_score}")

In production search code, we apply similar logic to each candidate product result, computing its final ranking score. This typically happens in-memory via custom comparators or by incorporating the weight directly into the scoring formula. If using Solr/Elasticsearch features, you can also leverage function queries or scripts (Solr's if function or Elasticsearch's script score) to include store ratings in the score calculation. In our implementation, we chose to perform calculation at the search service layer for directness and fine-grained control.

With this weighting strategy, search results suppress products from low-rating stores, raising the overall quality of products visible on early result pages. Specific thresholds and penalty ratios can be adjusted based on observed results. In the future, if we wish to boost high-rating stores, we could apply similar logic—for example, increase scores by a percentage for stores above 4.5—to incentivize seller excellence.

Outcome

Through this approach, we introduced store rating as a ranking dimension in the e-commerce search engine. The data synchronization, index update, and ranking adjustment mechanism successfully reduced low-rating stores' product visibility and improved result reliability and user experience. We ensured data timeliness and accuracy through combined real-time Kafka messaging and periodic batch processing, added index fields to support complex ranking logic, and designed the system as configurable and extensible so operations teams can adjust thresholds and weights according to business needs.

After deployment, the store rating weighting mechanism effectively reduced cases where low-reputation stores dominated early search results. We also observed improvements in search result click-through and conversion rates. This work demonstrates that incorporating business quality metrics—such as store ratings—into search ranking is an important way to improve result relevance and user satisfaction. Future work can combine additional signals like merchant response time or inventory turnover to further refine the ranking strategy and continue optimizing the e-commerce search experience.