DEV Community: Vladyslav Hutov

JavaScript Lenses: A Guide to Efficiently Modifying Immutable Data

Vladyslav Hutov — Mon, 30 Jan 2023 16:52:57 +0000

The software industry has experienced a paradigm shift towards functional programming in recent years.

A paradigm shift in programming refers to a change in the way we approach writing code. Contrary to the common belief, a paradigm doesn’t bring new features, it comes with a set of limitations. These limitations, when embraced, can bring benefits of writing better software.

Here is what I mean:

Assembly language is a low-level language with no limitations, allowing for unlimited access to memory locations and the ability to manipulate bytes, pointers, and code flows.
Structural programming introduced the concept of code blocks, with limitations on how we treat iterations and subroutines.
Object-Oriented Programming (OOP) added further limitations by controlling how internal memory of objects can be accessed through encapsulation and abstraction.
Functional programming, limits us to the use of pure functions and immutable structures, leading to greater clarity and understanding in the code. This is particularly useful in parallel and concurrent programming, as pure functions and immutable objects minimize the chance of unintended modifications.

Working with immutable structures in functional programming can present challenges, particularly in JavaScript which lacks intrinsic tools for immutability. This requires developers to take responsibility for ensuring immutability. Simple updates, such as changing a value through destructuring, are straightforward:

const person = {
  name: 'John',
  age: 25
}

const newPerson = { ...person, age: 26 }

Updating an array is similarly simple:

const users = [...]

const newUsers = users.concat([newPerson]) // or [...users, newPerson]

However, as the complexity increases with nested structures, it becomes more difficult to create a new copy with updated data. For example, updating a nested object like person.address:

const person = {
  name: 'John',
  address: {
    street: '1 Water Ln',
    city: 'London'
    index: 'NW1 8NX'
  }
}

const newPerson = {
  ...person,
  address: {
    ...person.address,
    index: 'NW1 8NZ'
 }
}

Or updating an object within an array:

const users = [...]

const newUsers = [
  {
    ...users[0],
    address: {
       ...users[0].address,
       index: 'NW1 8NZ'
  },
  ...users.slice(1)
]

The more nested the structures, the greater the challenge in making updates without mutating the original data.

JavaScript doesn’t support an easy way to modify immutable data. - maybe YOU thinking

What is the solution then? Should we discard the concepts of pure functions and immutable structures? Should we avoid nesting objects and arrays? Is it necessary to switch to a different programming language?

The answer to each of these is no. Today, I'd like to introduce a concept that can address this issue. This concept emerged in 2010 with the extension library Lens for the Haskell programming language.

Don't be intimidated by Haskell, you'll realise the practicality of the Lens concept in a moment - me

A Lens allows you to focus on a specific part of an immutable data structure, serving as a functional equivalent of a getter and setter. It can be described as a generic class with two functions: a getter and a setter.

class Lens<S, A> {
  getter: (S) => A
  setter: (A, S) => S
}

Where S represents an immutable object and A is the focused value within that object. The getter function defines how to view the value, and the setter is a pure function for updating the object and creating a new instance of S.

Why S and A? I couldn’t find, however, I can speculate that S stands for "Source" and A stands for "Focus". The letter F is often reserved for representing higher-order types, so A was chosen as a common placeholder for simple types.

The implementation of lenses in JavaScript can be done using libraries like Ramda, which provides functions for constructing lenses on objects and arrays. Ramda's R.lens function creates a lens by providing the getter and setter functions, while R.lensProp and R.lensIndex functions offer shortcuts for commonly used getters and setters.

const nameLens = R.lens(/*getter*/R.prop('name'), /*setter*/R.assoc('name'))
// use of shortcuts
const nameLens = R.lensProp('name')
const firstLens = R.lensIndex(0)

So now that we have lenses, what can we do with them? Lenses provide us with the ability to retrieve (using R.view), modify (using R.set), or update (using R.over) properties within a structure.

const persons = [
  {name: 'John'},
  {name: 'Steve'}
]

const updateName = person =>
  R.over(nameLens, name => name + ' Smith', person)

R.over(firstLens, updateName, persons) // [{"name": "John Smith"}, {"name": "Steve"}]

persons // [{"name": "John"}, {"name": "Steve"}]

In the example above, we utilised an intermediate function to update an array's value. But did you know that lenses can be composed? Let's write a helper function to combine two lenses, allowing us to focus on a smaller part of a structure with a single lens.

function andThen<S, S1, A>(lens1: Lens[S, S1], lens2: Lens[S1, A]): Lens[S, A] {

  const getter = R.pipe(
    R.view(lens1),
    R.view(lens2)
  )

  const setter = (a, s) => 
    R.over(lens1, R.set(lens2, a), s)

  return R.lens(getter, setter)
}

This is how we can use the function:

const nameLens = R.lensProp('name')
const firstLens = R.lensIndex(0)

const nameInFirstLens = andThen(firstLens, nameLens)

const persons = [
  {name: 'John'},
  {name: 'Steve'}
]

R.over(nameInFirstLens, name => name + ' Smith', persons) // [{"name": "John Smith"}, {"name": "Steve"}]
persons // [{"name": "John"}, {"name": "Steve"}]

Let's break down how the andThen composition works.

We'll start with the getter aspect. We have a top-level lens lens1 that focuses on the internal structure S1 of S and another lens lens2 that focuses on the internal value A of S1.

Here's how the getter is implemented:

const getter = R.pipe(
   R.view(lens1),
   R.view(lens2)
 )

R.pipe runs the functions in sequence, with the output from one function serving as input to the next. In this case, the getter first uses lens1 to view the value, which gives the internal structure, then uses lens2 to view the value within that structure.

The setter implementation is more complex:

const setter = (a, s) => 
    R.over(lens1, R.set(lens2, a), s)

The setter is a function that takes two inputs (A, S) and returns S. Let's start by examining R.set(lens2, a), which creates an update function S1 => S1 that updates the value focused by lens2 to a.

The R.over function then updates the value S1 in S using lens1 to focus on it and the update function created by R.set. The signature of R.over is <S, S1>(Lens[S, S1>, S1 => S1, S) => S.

You can simplify lens composition with R.compose from Ramda:

const nameLens = R.lensProp('name')
const firstLens = R.lensIndex(0)

const nameInFirstLens = R.compose(firstLens, nameLens)

There is a difference in the lens composition approach described with andThen and the one with R.compose, but the end result is the same and there's no need to worry about it.

For object-specific composition you can use R.lensPath:

const addressLens = R.lensProp('address')
const zipLens = R.lensProp('zip')

const zipInAddressLens = R.compose(addressLens, zipLens)
const zipInAddressLens2 = R.lensPath(['address', 'zip'])

Multiple updates can be composed as well:

const nameLens = R.lensProp('name')
const ageLens = R.lensProp('age')
const zipAddressLens = R.lensPath(['address', 'zip'])

const firstLens = R.lensIndex(0)

const nameInFirst = R.compose(firstLens, nameLens)
const ageInFirst = R.compose(firstLens, ageLens)
const zipInFirst = R.compose(firstLens, zipAddressLens)

const persons = [
  {name: 'John', age: 25, address: { zip: 'NW1 8NZ' } },
  {name: 'Steve', age: 35, address: { zip: 'NW1 8NZ' } }
]

R.compose(
  R.set(ageInFirst, 30),
  R.over(nameInFirst, name => name + ' Smith'),
  R.set(zipInFirst, 'NW1 8NX')
)(persons)
/*
  [
    {
        address: {
            zip: "NW1 8NX"
        },
        age: 30,
        name: "John Smith"
    },
    {
        address: {
            zip: "NW1 8NZ"
        },
        age: 35,
        name: "Steve"
    }
]
*/

However, this approach duplicates the array three times to update one person. To reduce this, you can create a composed update function for a single person:

const updatePerson = R.compose(
  R.set(ageLens, 30),
  R.over(nameLens, name => name + ' Smith'),
  R.set(zipAddressLens, 'NW1 8NX')
)

And then update the array using the lens which focuses on specific person:

R.over(firstLens, updatePerson, persons)

This produces the same result but avoids creating intermediate arrays that are later discarded.

Now you have a tool to immutably update objects in JS of any composition structure without boilerplate.

If you have questions, leave your comments below!

Follow for more!

Building Real-Time Recommendation System

Vladyslav Hutov — Mon, 23 Jan 2023 11:25:51 +0000

In the first part of the series on building a personalised music recommendation system, we discussed the use of machine learning algorithms such as Collaborative Filtering and Co-occurrence Counting to create an items similarity matrix. This resulted in a set of files, each representing the similarities between various artists and tracks.
As an example, one of these files may look like this:

00FQb4jTyendYWaN8pK0wa 0C8ZW7ezQVs4URX5aX7Kqx ... 66CXWjxzNUsdJxJ2JdwvnR
02kJSzxNuaWGqwubyUba0Z 137W8MRPWKqSmrBGDBFSop ... 3nFkdlSjzX9mRTtwJOzDYB
04gDigrS5kc9YWfZHwBETP 0du5cEVh5yTK9QJze8zA0C ... 1Xyo4u8uXC1ZmMpatF05PJ
06HL4z0CvFAxyc27GXpf02 4AK6F7OLvEQ5QYCBNiQWHq ... 26VFTg2z8YR0cCuwLzESi2
07YZf4WDAMNwqr4jfgOZ8y 23zg3TcAtWQy7J6upgbUnj ... 6LuN9FCkKOj5PcnpouEgny

Here, the first entry is the key entity, followed by N entities that are similar to it.

In this second part of the series, we will delve into the process of building a Node.js application that utilises this data and other information to provide real-time music recommendations to users, based on their individual tastes. We will explore the use of different database paradigms to ensure our solution is both scalable and performant.

If you're just joining and want to follow along as we build this system, but don't want to dive into the world of machine learning, you'll need the following data:

The original raw playlist dataset, which can be found at Kaggle.
The similarity matrices, which can be found at GitHub, and contain artist similarities (matrix factorization, MLP, and co-occurrence) and track similarities (matrix factorization).

With that out of the way, let's get started!

System Architecture

The Node.js application we are building will consist of three peripheral services and a high-level recommendation service with a fluent API, a request handler, and an express route to handle requests.

The following diagram gives a high-level overview of the application architecture:

To build this app, we will start by implementing the lower-level services before moving on to the higher-level ones. This approach is known as building bottom-up, and I hope it will help understand the process better.

Services

TracksService

To store and retrieve the track and artist data, we will use a relational database paradigm. This requires us to create two tables: one for storing artists, and another for storing the tracks of each artist.

Here is the SQL code needed to create these tables:

CREATE TABLE artists (
  uri varchar(32) NOT NULL,
  name varchar(100) NOT NULL,
  PRIMARY KEY (uri)
);

CREATE TABLE tracks (
  uri varchar(32) NOT NULL,
  name varchar(100) NOT NULL,
  artist_uri varchar(32) NOT NULL,
  PRIMARY KEY (uri),
  KEY artist_uri (artist_uri),
  CONSTRAINT tracks_fk_1 FOREIGN KEY (artist_uri) REFERENCES artists (uri)
);

To run all the services we need, we will use Docker containers. One option is to use docker-compose with the mysql image to create these tables and store the data persistently.

Here is an example of a docker-compose.yml file that can be used to create a MySQL container and set up the necessary environment variables, volumes, and ports:

services:
  mysql-server:
    image: mysql:8.0.31
    command: --default-authentication-plugin=mysql_native_password
    environment:
      MYSQL_ROOT_PASSWORD: admin123
      MYSQL_DATABASE: music
      MYSQL_USER: user
      MYSQL_PASSWORD: user123
    volumes:
      - db:/var/lib/mysql
    ports:
      - 3306:3306
volumes:
  db:

You need to have a docker-compose installed on your machine.

Using the above docker-compose.yml file, we can easily start a MySQL server and connect to it from our Node.js application.

To populate the data from the raw playlist dataset into the database, we will need to write a script. I used Python for this task as it is simple to use, but since the main focus of this article is the Node.js application for recommendations, I will not go into the details of how the script works. You can find the script in the GitHub repo.

You can run this script to import the data from the raw dataset into the database. Once the data is imported, our application can then use this data to query tracks and artists.

Now, let's implement the TracksService class. This service will be responsible for handling various queries related to track data, this is the service interface:

class TrackService {
    /** @type {knex.Knex} */
    #db

    constructor(db) {
        this.#db = db
    }

    asyncTrackData = async (trackIds) => { ... }
    asyncGetTracksByArtist = async (artistIds, {fanOut, randomize}) => { ... }
}

The first function is a simple query for track data. It selects tracks by their URI and renames the uri column to id, which we will need later on.

/**
 * Gets track data from db
 *
 * @param {string[]} trackIds
 *
 * @returns {Promise<Object.<string, any>[]>} tracks data
 */
asyncTrackData = async (trackIds) => {
    const rows = await this.#db
        .select()
        .from('tracks')
        .whereIn('uri', trackIds)

    return rows.map(({ uri: id, ...rest }) => ({ id, ...rest }))
}

This function accepts an array of track IDs as an input and returns a promise that resolves to an array of track data. It uses the knex.js library to connect and query the MySQL database.

The second function of the TracksService class allows us to query track IDs by their authors. It may appear complex at first glance, but the complexity arises from the fact that we want to limit the maximum number of tracks per author.

/**
 * Get list of tracks per artist.
 *
 * @param {string[]} artistIds list of artist ids
 * @param {Object} options configuration:
 * @param {number} [options.fanOut] max number tracks per artist
 * @param {boolean} [options.randomize = false] when fan_out specified - if true, will randomly shuffle tracks before limiting
 *
 * @returns {Promise<Object.<string, string[]>>} map where key is artist_id and value is a list of artist track_ids
 */
asyncGetTracksByArtist = async (artistIds, { fanOut = null, randomize = false } = {}) => {
    const rows = await this.#db
        .select({
            track_id: 'uri',
            artist_id: 'artist_uri'
        })
        .from('tracks')
        .whereIn('artist_uri', artistIds)

    const fanOutFun = 
        randomize ? 
            (v) => _.sampleSize(v, fanOut) : 
            (v) => _.slice(v, 0, fanOut)

    const applyLimitToArtist = R.map(([k, v]) => [k, fanOutFun(v)])

    const groupByArtist = R.groupBy(R.prop('artist_id'))

    const limitPerArtist = fanOut
        ? R.pipe(
              R.toPairs,          // split object to a list of tuples
              applyLimitToArtist, // per tupple apply the limit function
              R.fromPairs         // construct the object back
          )
        : R.identity              // if fanout is false, do nothing

    /*
      For each value within parent object, take field track_id
      Convert 
      {
        artist_id1: [{ track_id: a, artist_id: artist_id1}],
        artist_id2: [{ track_id: b, artist_id: artist_id2}]
      }
      to
      {
        artist_id1: [a],
        artist_id2: [b]
      }
    */
    const projectTrackId = R.mapObjIndexed(R.pluck('track_id'))

    return R.pipe(groupByArtist, limitPerArtist, projectTrackId)(rows)
}

It starts by querying an unlimited number of tracks per artist, and then, depending on the input arguments, randomly or deterministically limits the number of tracks. We make use of two additional libraries here: Ramda and Lodash for object and array manipulations.

It's important to note that artist_uri column in the tracks table is a foreign key, which means that MySQL creates a secondary index on this field. Having a secondary index ensures that our query will run quickly even when we have a large amount of data in the database, as we do not need to perform a full scan of the table.

SimilarityService

Our next service helps us to find similar entities. As you may recall, we previously created similarity matrices, which hold the similarity relations.

The service API is straightforward: given a list of entity IDs and an index (which serves as an identifier for the ML model used to build a similarity matrix), the function returns a list of similar entities.

class SimilarityService {
    /** @type {redis.RedisClientType} */
    #redis

    constructor(redis) {
        this.#redis = redis
    }

    asyncGetSimilar = async (ids, { indexName, fan_out }) => { ... }
}

While it would be possible to use SQL to model this functionality, I propose using a key-value based NoSQL solution. We can use an in-memory database, such as Redis, to store our similarity matrices, as they can fit in RAM. For example, the artist files use less than 500 KB and the track files need 2 MB. Even if we had 1,000 or 10,000 more artists (and tracks), it would still fit within memory. Additionally, we can also shard if more memory is needed.
Using RAM based storage type we ensure that request is fulfilled with the least possible delay. Redis also has persistence mode, which we will use for durability of the data.

To use our similarity service, we will need to update our docker-compose file to start a Redis server as well. Here is an example of how we can do that:

services:
  redis:
    image: redis
    ports:
      - 6379:6379
    volumes:
      - redis:/data
    command: redis-server --save 60 1 --loglevel warning
volumes:
  redis:

The schema we will use for storing the similarity data in Redis is as follows:

key: list[entity_id]

The key consists of two parts: <index_name>:<entity_id>.

We will also need a script that populates Redis with similarity data. This script can be found here. Once the script is run, it will create four different similarity indices.

With the setup in place, we can now write the implementation for the similarity service:

/**
 * For provided entity ids fetches similar entities
 *
 * @param {string[]} ids
 * @param {Object} options
 * @param {string} options.indexName redis index name
 * @param {number} [options.fanOut = 10] limit number of similar entities per entity
 *
 * @returns {Promise<Object.<string, string[]>>} map of similar entities, where key is input entity_id and values are similar entity_ids
 */
asyncGetSimilar = async (ids, { indexName, fanOut = 10 }) => {
    const key = (id) => `${indexName}:${id}`

    // creates array of promises
    const pendingSimilarities = ids.map(async (id) => {
        const similarIds = await this.#redis.lRange(key(id), 0, fanOut)

        if (similarIds.length == 0) return null
        return [id, similarIds]
    })

    // when promises are awaited, we get a list of tuples [id, similarIds]
    // ideally we want to have some error handling here
    const similarities = (await Promise.allSettled(pendingSimilarities)).filter((r) => r.value).map((r) => r.value)

    return Object.fromEntries(similarities)
}

The implementation is simple - it executes N concurrent requests to Redis to get array slices for the provided entity keys. It returns an object, where the key is the original entity ID and the value is an array of similar entity IDs.
Ideally we should have error handling for the promises here to handle any errors that might occur when querying Redis.

FuzzySearch

Our recommendation system is not based on user activities, and we do not have access to internal track IDs. Instead, we will be relying on arbitrary strings of text entered by the user to specify track or artist names. In order to provide accurate recommendations, we will need to implement a fuzzy search mechanism for tracks and artists.

Fuzzy search is an approach that can help us build a robust system for finding actual tracks based on user input. It allows for a degree of flexibility in the user's search queries, making it less prone to errors. A standard example of a system which supports such queries is ElasticSearch.

To begin, let's take a look at our docker-compose configuration for setting up ElasticSearch:

services:
  elastic:
    image: elasticsearch:8.6.0
    ports:
      - 9200:9200
      - 9300:9300
    environment:
      - xpack.security.enabled=false
      - discovery.type=single-node
    volumes:
      - elastic:/usr/share/elasticsearch/data
volumes:
  elastic:

In our implementation, we will be using two types of indices in ElasticSearch: one for indexing tracks with artists, and another for indexing just artists. This is similar to a reverse representation of the original SQL model.

The diagram below illustrates how we are storing and querying data in SQL:

And this is how we are storing them in ElasticSearch:

The ElasticSearch indices allow us to query the required track and artist IDs by their names very quickly, providing a fast and efficient search process. This improves the overall user experience, as the system is able to provide recommendations in a timely manner.

The built-in fuzzy search mechanism enables us to handle variations in the user's input, such as misspellings or slight variations in the search query. This ensures that our system can still provide accurate recommendations even if the user's input is not perfect.

The FuzzySearch class is responsible for handling the search functionality in our recommendation system:

class FuzzySearch {
    /**
     * @type {Client}
     */
    #es

    constructor(es) {
        this.#es = es
    }

    asyncSearchArtist = async (artists) => { ... }  
    asyncSearchTrack = async (tracks) => { ... }
}

The asyncSearchArtists method is responsible for searching for artists in the "artists" index using the provided name(s) as the search query. The method uses a helper method called matchField to compose the "SELECT" part of the ElasticSearch query.

/**
 * @param {string} fieldName field name which need to match
 * @param {string} value matching value
 * @returns elasticsearch query for matching <fieldName> field
 */
const matchField = (fieldName, value) => ({
    match: {
        [fieldName]: {
            query: value,
            operator: 'AND'
        }
    }
})

The asyncSearchArtists method then constructs the full request, specifying the "artists" index and the query for each artist. It uses the ElasticSearch client's msearch (multiple search) method to perform the search:

/**
 * @param {{artist: string}[]} artists 
 *        array of query strings for matching artists 
 *        each object must contain artist name
 * @returns {Promise<SearchResult[]>} track and artist ids
 */
asyncSearchArtists = async (artists) => {
    const { responses } = await this.#es.msearch({
        searches: artists.flatMap(({ artist }) => [
            { index: 'artists' },
            {
                query: matchField('name', artist)
            }
        ])
    })

    return maybeGetFirstFrom(responses)
}

The search results returned by ElasticSearch may contain zero to many matched objects, so we need to flatten the result set. For simplicity, let’s select the first result from the result set if it exists.

const maybeGetFirstFrom = (responses) =>
    responses.flatMap((r) => {
        if (!r.hits.hits) return []

        const { _id } = r.hits.hits[0]

        return [
            {
                id: _id
            }
        ]
    })

When searching for tracks, we have two cases: when artist name search clause is present or is missing, so let’s create a helper function for constructing the query:

/**
 * @param {{track: string, artist: ?string}} value matching value 
 * @returns {{match: Object}[]} array of matching clauses in elastic query dsl
 */
const searchTrackTerms = ({ track, artist = null }) => {
    const trackTerm = matchField('name', track)

    if (!artist) {
        return [trackTerm]
    }

    const artistTerm = matchField('artist', artist)

    return [trackTerm, artistTerm]
}

The asyncSearchTracks method makes use of this helper function to construct the query for searching for tracks in the "tracks" index.

/**
 * @param {{track: string, artist: ?string}[]} tracks 
 *        array of query strings for matching tracks 
 *        each object must contain either track name or track and artist names
 *        having artist name present increases likelyhood of finding the right track
 * @returns {Promise<{id: string}[]>} track ids for matched queries
 */
asyncSearchTracks = async (tracks) => {
    const { responses } = await this.#es.msearch({
        searches: tracks.flatMap((track) => [
            { index: 'tracks' },
            {
                query: {
                    bool: {
                        must: searchTrackTerms(track)
                    }
                }
            }
        ])
    })

    return maybeGetFirstFrom(responses)
}

This concludes the implementation of peripheral services. In summary, we have implemented three services which we’ll use to build our recommendation system:

Data retrieval using an SQL model: This service allows us to quickly lookup data by its ID, providing fast access to the required information.
Similarity retrieval using an in-memory key-value model: This service provides a blazing fast access to the similarity matrix, which is crucial for generating recommendations based on similar items.
Fuzzy search using a text index: This service allows us to quickly find relevant entity IDs, even when the user's input is not perfect. The built-in search mechanism in the text index provides a degree of flexibility in the search process, handling variations in the user's input.

Each of these services targets to decrease latency time in their respective zones of responsibility. Together, they provide a robust and efficient ground for the recommendation system that can handle variations in user input, while providing accurate and timely recommendations.

Fluent API

In the remaining sections, we will look at how we can integrate the services we have built to create a functional recommendation system. While we could start using the services as-is, building a real-world recommendation flow would become tedious and cumbersome.

To improve the development process, we need to build a data model and a service that provides a fluent API for construction recommendation flows.

The core concepts in this data model will be a dynamic entity and a pipe. A dynamic entity represents a piece of data that can change over time, such as a user's query or a track's artist. A pipe represents a flow of data through the system, allowing us to apply different operations on the data as it moves through the system.

The data model and service will provide a simple and intuitive way to build recommendation systems, making it easy to experiment with different approaches and fine-tune the system to achieve the desired performance.

Entity

As mentioned previously, a dynamic entity is just a container of properties. It is a plain JavaScript object that can hold any properties that are relevant to the recommendation system.

We won't enforce any specific structure for this entity, and the responsibility of type checking will be on the client of this API. This allows for maximum flexibility in the data model, and makes it easy to add or remove properties as needed.

Pipe

A pipe is an asynchronous function that receives an array of entities and returns a modified array of entities. It can perform various operations on them, such as filtering, sorting, or transforming the data. For example, a pipe can filter out entities that do not meet certain criteria, sort based on a specific property, or transform the entities by adding or removing properties.

If a pipe creates a new entity, it copies over properties from the parent entity. This allows for the preservation of relevant information from the original entity, while also allowing for the addition of new properties.

RecommendationService

With the concepts of dynamic entities and pipes, we can now introduce the RecommendationService API.

type Config = Object
type Entity = { [string]: any }
type OneOrMany<A> = A | A[]
type Pipe = (OneOrMany<Entity> => Promise<Entity[]>)


class RecommendationService {
    /* Peripheral API */
    fuzzySearchTracks: Pipe
    fuzzySearchArtists: Pipe
    enrichTrack: Pipe
    similar: (options: Config) => Pipe
    artistTracks: (options: Config) => Pipe

    /* Util API */
    dedupe: (by: string | string[]) => Pipe
    diversify: (by: string | string[]) => Pipe
    sort: (by: string | string[]) => Pipe
    take: (limit: int) => Pipe
    set: (from: string, to: string) => Pipe
    setVal: (key: string, value: any) => Pipe

    /* Composition API */
    merge: (...pipes: Pipe[]) => Pipe
    compose: (...pipes: Pipe[]) => Pipe

    /* internals */
    #similarityService: SimilarityService
    #trackService: TrackService
    #fuzzyService: FuzzySearch
}

The API is split into three main sections: peripheral API, util API and composition API.

Peripheral API provides the core services that are necessary to build a recommendation system, such as the fuzzySearchTracks, fuzzySearchArtists, enrichTrack, similar and artistTracks methods. These methods provide access to the services that were previously created and make it easy to retrieve data from the system.

Util API provides utility methods for manipulating the data as it moves through the system, such as the dedupe, diversify, sort, take, set, and setVal methods.

Composition API provides methods for composing and merging pipes. These methods allow for the creation of complex data manipulation flows by combining multiple individual pipe functions.

The full implementation of the RecommendationService service is not included in the article, but it can be found in the GitHub repository.

Let’s take a look at the fuzzySearchTracks method as an example of how we can implement one of the peripheral API methods in the RecommendationService. The method takes an input of user search requests, which are represented as dynamic entities. Each entity must contain a track name and optionally an artist name:

/**
 * @typedef {Object.<string, any>} Entities
 * @param {Entity|Entity[]} input user search requests. Each entity must contain track name and optionally artist name
 * @returns {Promise.<Entity[]>} found track ids
 */
fuzzySearchTracks = async (input) => {
    input = toArray(input)

    const trackInputs = input.filter((v) => v.track)

    const searchResults = await this.#fuzzyService.asyncSearchTracks(trackInputs)

    return searchResults
}

The fuzzySearchTracks method does not preserve any of the parent properties of the input entities, as they have no use in the recommendation flow. The method serves as an entry point and its primary function is to retrieve the relevant track ids.

The similar method is another example of how we can implement one of the peripheral API methods. It takes an input of entities and an options object, and finds similar entities.

/**
 * Finds similar entities. Copies properties from parent to children.
 * @typedef {Object.<string, any>} Entities
 * @param {Object|Function} options see SimilarityService#asyncGetSimilar options
 * @param {Entity|Entity[]} input
 * @returns {Promise<Entity[]>} similar entities for every entity from input
 */
similar = (options) => async (input) => {
    input = toArray(input)

    options = _.isFunction(options) ? options() : options

    const ids = input.map(R.prop('id'))

    const similarMap = await this.#similarityService.asyncGetSimilar(ids, options)

    return input.flatMap((entity) => {
        const similar = similarMap[entity.id] || []

        return similar.map((id) => ({
            ...entity,
            id
        }))
    })
}

The method then maps over the input entities and for each entity, it flattens an array of similar entities by copying over the properties from the parent entity to the child entities, which get new ids. This allows for the preservation of relevant information from the original entity.

Let’s also overview how composition functions work. We have two of those: merge and compose.

The merge takes a list of pipes and runs them in parallel, creating a higher-level pipe. The high-level pipe internally feeds input to the composed parts, runs them concurrently, and then merges their output.

/**
 * Merges few pipes into single pipe
 * @typedef {Object.<string, any>} Entities
 * @param  {...(Entity[] => Promise<Entity[]>)} pipes
 * @returns high level pipe
 */
merge = (...pipes) => async (input) => {

    // converging function receives an array of triggered promisses
    // and awaits all of them concurrently
    const convergingF = async (...flows) =>     
        (await Promise.allSettled(flows))
            .filter((v) => v.value) // missing error handling
            .map((v) => v.value)
            .flat()


    return R.converge(convergingF, pipes)(input)

}

The merge function is a powerful tool that allows us to run multiple pipes concurrently, which can greatly speed up the process of generating recommendations.

The compose function is another composition function in the RecommendationService. It creates a sequential execution of pipes, where the output of one pipe is fed as input to the next pipe:

/**
 * Creates sequential composition of pipes
 * @typedef {Object.<string, any>} Entities
 * @param  {...(Entity[] => Promise<Entity[]>)} pipes
 * @returns high level pipe
 */
compose = (...pipes) => async (input) => {
    return R.pipeWith(R.andThen, pipes)(input)
}

Using the compose function we can create complex recommendation flows.

FlowBuilder

We can now utilise this API to construct a recommendation flow. The flow will be split into two branches: we will be recommending tracks based on the user's track preferences and also based on their artist preferences, as seen in the flow diagram.

The following is the interface for the flow builder:

const buildFuzzySearchFlow = (config, recs: RecommendationService) => {
    // Importing the fluent API interface
    const {
        fuzzySearchTracks,
        fuzzySearchArtists,
        similar,
        enrichTrack,
        artistTracks,
        dedupe,
        diversify,
        take,
        set,
        setVal,
        merge,
        compose
    } = recs

    // Flow implementation
    ...
}

And this is how we can implement the flow based on user search input:

const artistFlow = compose(
    dedupe('artist'),
    fuzzySearchArtists,
    dedupe('id'),
    set('id', 'recommender'),
    setVal('flow', 'artist-flow'),
    similar(config.recs.artist.mlp),
    dedupe('id'),
    artistTracks(config.artistTracks),
    diversify(['recommender', 'artist_id']),
    take(50)
)

const trackFlow = compose(
    fuzzySearchTracks,
    dedupe('id'),
    set('id', 'recommender'), 
    setVal('flow', 'track-flow'),
    similar(config.recs.track),
    dedupe('id'),
    diversify('recommender'),
    take(50)
)

return compose(
    merge(artistFlow, trackFlow),
    dedupe('id'),
    diversify('flow'),
    take(config.limit),
    enrichTrack
)

Let’s dive into the specifics of the artist recommendations branch.

To begin, we use the compose function to build a sequential chain of underlying pipes. This allows us to perform a series of operations in a specific order.

First, we use the dedupe function to take unique artist names (based on the user's search input) and then query ElasticSearch to retrieve artist IDs. Then, we use the dedupe function again to ensure that the results from ElasticSearch are also unique.

dedupe('artist')
fuzzySearchArtists
dedupe('id')

Next, we use the set and setVal functions to create new properties on each entity.

set('id', 'recommender')
setVal('flow', 'artist-flow')

The set function creates a new property called "recommender" by copying the value of the entity's "id" property. The setVal function creates a new property called "flow" with a constant value of "artist-flow". These two functions will allow us to trace the origin of recommendations later on when needed.

We then move on to finding similar artists to the ones provided by the user.

similar(config.recs.artist.mlp)
dedupe('id')

This is done by querying the similar index with values specified in the config file. We then use the dedupe function again to remove any duplicate results.

Finally, the last step in the artist flow is to retrieve the songs for the recommended artists and limit the number of results.

artistTracks(config.artistTracks)
diversify(['recommender', 'artist_id'])
take(50)

To ensure a more natural shuffle of the results, the diversify function is used, which uses a Round-Robin shuffling mechanism:

When we call the full flow, we will receive results similar to the example shown below.

[
  {
    id: '1VpSH1BPdKa7KYVjH1O892',
    recommender: '711MCceyCBcFnzjGY4Q7Un',
    flow: 'artist-flow',
    artist_id: '4F84IBURUo98rz4r61KF70',
    name: 'The Air Near My Fingers',
    artist_uri: '4F84IBURUo98rz4r61KF70'
  },
  {
    id: '32lm3769IRfcnrQV11LO4E',
    flow: 'track-flow',
    recommender: '08mG3Y1vljYA6bvDt4Wqkj',
    name: 'Bailando - Spanish Version',
    artist_uri: '7qG3b048QCHVRO5Pv1T5lw'
  },
  {
    id: '76wJIkA63AgwA92hUhpE2V',
    recommender: '711MCceyCBcFnzjGY4Q7Un',
    flow: 'artist-flow',
    artist_id: '1ZwdS5xdxEREPySFridCfh',
    name: 'Me Against The World',
    artist_uri: '1ZwdS5xdxEREPySFridCfh'
  },
  ...
]

Each result will have properties such as "id", "recommender", "flow", "artist_id", "name", and "artist_uri". These properties provide information about the recommended track, as well as where it came from in the recommendation flow.

The full code, that includes a request handler and express app, can be found in the GitHub repository.

That's it. I hope you have enjoyed building the recommendation systems and learned something new!

Building Personalised Music Recommendation System

Vladyslav Hutov — Wed, 11 Jan 2023 14:31:41 +0000

I was excited to find a playlist dataset from Spotify because I have always been unhappy with the recommendations from music streaming platforms. I wanted to use this dataset as an opportunity to learn how to build a personalised recommendation system.

In this series, I'll walk you through the process of building such a system using Python for data processing and Node.js for recommendations retrieval. In this first article, we'll focus on the data processing flow. In the second article, we'll be building a real-time recommendation system.
If you're interested, join me to see the final result!

Conceptually, this is what I wanted to implement:

The Data

The dataset we'll be working with consists of 1000 JSON files, and contains around 1 million public playlists.

Here's an example of a single playlist from the JSON:
playlist content

{
  "name": "Throwbacks",
  "collaborative": "false",
  "pid": 0,
  "modified_at": 1493424000,
  "num_tracks": 52,
  "num_albums": 47,
  "num_followers": 1,
  "tracks": [
    {
      "pos": 0,
      "artist_name": "Missy Elliott",
      "track_uri": "spotify:track:0UaMYEvWZi0ZqiDOoHU3YI",
      "artist_uri": "spotify:artist:2wIVse2owClT7go1WT98tk",
      "track_name": "Lose Control (feat. Ciara & Fat Man Scoop)",
      "album_uri": "spotify:album:6vV5UrXcfyQD1wu4Qo2I9K",
      "duration_ms": 226863,
      "album_name": "The Cookbook"
    },
    // ...
    {
      "pos": 51,
      "artist_name": "Boys Like Girls",
      "track_uri": "spotify:track:6GIrIt2M39wEGwjCQjGChX",
      "artist_uri": "spotify:artist:0vWCyXMrrvMlCcepuOJaGI",
      "track_name": "The Great Escape",
      "album_uri": "spotify:album:4WqgusSAgXkrjbXzqdBY68",
      "duration_ms": 206520,
      "album_name": "Boys Like Girls"
    }
  ],
  "num_edits": 6,
  "duration_ms": 11532414,
  "num_artists": 37
}

Each playlist consists of a name, a PID, and a list of tracks, along with some additional data.

When building this recommendation system, I made the assumption that "people are more likely to add similar songs in a playlist." While this may not hold true for every playlist, I believe that on a larger scale, it's a reasonable assumption. As a result, I'll be treating each playlist (identified by its PID) as a unit of taste preferences, and modeling item similarities based on them using various algorithms.

Talk is cheap. Show me the code ― Linus Torvalds

Preparing Data

Before diving into running different ML algorithms, we need to prepare the data for ingestion and do some cleaning.

The goal is to parse the URIs, select the necessary fields, unfold the nested playlist tracks structure, and keep only the top 10K artists. Also, converting the JSON files to CSV and parsing the fields will save time when experimenting.

While this section may be basic, I've included it for completeness. If you're familiar with these concepts, feel free to skip ahead to the next section.

Below are some common imports that I'll be using throughout the examples.

imports

import collections
import functools
import itertools
import json
import math

from collections import Counter, namedtuple
from pathlib import Path

import pandas as pd
import tqdm

from fastai.collab import *
from fastai.tabular.all import *
from sklearn.neighbors import KNeighborsClassifier

Reading JSON dataset

We'll need a few helper functions. The first one is simply reading a file and parsing the JSON contained within it:

def read_file(path):
    with open(path, 'r') as f:
      return json.load(f)

We'll also need a few field getters for the track entity. Notice that the URI parsing is just a simple substring operation:

artist_uri_prefix = len('spotify:artist:')
track_uri_prefix = len('spotify:track:')


def artist_name(track):
    return track['artist_name']

def artist_uri(track):
    return track['artist_uri'][artist_uri_prefix:]

def track_name(track):
    return track['track_name']

def track_uri(track):
    return track['track_uri'][track_uri_prefix:]

The next function reads the JSON files, parses the required fields, and returns an iterator of processed rows. This part might be confusing if you're not familiar with the concept of generators. Since we don't need to store the intermediate JSON files or the unfolded structure in memory, I'm using generators to lazily process them.

root_dir = '/kaggle/input/spotify-millions-playlist/spotify/data'
p = Path(root_dir)
paths = list(p.iterdir())

def read_files(num_files=5):

    lookup_path = paths if num_files is None else paths[:num_files]

    jsons = (read_file(path) for path in lookup_path)

    return (
        [
            row['pid'], 
            artist_name(track), 
            artist_uri(track), 
            track_name(track), 
            track_uri(track)
        ]
        for data in jsons
        for row in data['playlists']
        for track in row['tracks']
    )

example of a row

next(read_files(1))

[35000,
 'Marian Hill',
 '1xHQO9GJIW9OXHxGBISYc5',
 'Lovit',
 '2Ag3LUfgpN6ymEMwDOqKdg']

Most popular artists

Despite having a million of playlists, these data may not be enough to produce accurate predictions for every track and artist in the dataset. As a result, I want to limit the task to providing recommendations for the N most popular artists.

To do this, I'll need to count the number of artist mentions in the playlists:

artist_uri_pos = 2
c = collections.Counter(r[artist_uri_pos] for r in read_files(None))

After executing this code, I have the following structure:

c.most_common(1)

[('3TVXtAsR1Inumwj472S9r4', 846937)]

You can visit https://open.spotify.com/artist/3TVXtAsR1Inumwj472S9r4 to see the most popular artist according to the data we have from Spotify.

I should have actually counted unique artist mentions in a playlist here, but in this case it's not a problem since I'm not planning to use all 10K.

Let's save the top artists in a CSV for later use:

df = pd.DataFrame(data=c.most_common(10000), columns=['artist_uri', 'count'])
df.to_csv('common_artist_uri.csv', index=False)

Transforming JSON to CSV

Now, we want to convert the initial JSON files to processed CSV. To do this, we'll start by reading the most popular artists created in the previous step:

artists = pd.read_csv('/kaggle/input/common-spotify-artists/common_artist_uri.csv')
artists = artists.set_index('artist_uri')

I'll also modify the previously introduced parsing function by filtering out artists that are not in the allowlist and playlists that have too few tracks. This will help ensure that the models will produce better predictions:

def read_files(artist_allowlist, num_files=5):

    lookup_path = paths if num_files is None else paths[:num_files]

    jsons = (read_file(path) for path in lookup_path)

    return (
        [
            row['pid'], 
            artist_name(track), 
            artist_uri(track), 
            track_name(track), 
            track_uri(track)
        ]
        for data in jsons
        for row in data['playlists']
        if row['num_tracks'] > 5 # added filter or playlist number of tracks
        for track in row['tracks']
        if artist_uri(track) in artist_allowlist # and filter on popular artists
    )

Since we don't want to write everything in a single huge file, I'll split the rows into batches of 500K.

Here's the code for writing the CSV files:

gen = read_files(artists.index, None)
n = 500_000
i = 1
column_names = ['pid', 'artist_name', 'artist_uri', 'track_name', 'track_uri']

while (True):
    batch_slice = itertools.islice(gen, n)
    df = pd.DataFrame(batch_slice, columns=column_names)
    if df.empty:
        break
    df.to_csv(f'playlist_{i}.csv', index=False)
    i += 1

This leaves us with 5 GB of CSV files containing playlist-track data.

Similar items

In this section, we will compute the similarity between items in the dataset, such as artists or tracks. To do this, we will use three different algorithms: two variants of Collaborative Filtering and one based on Co-Occurrence.

Collaborative Filtering is a popular approach for recommendation systems, as it relies on the ratings or preferences of users to make predictions about what items a given user might like. In our case, we will use the presence of an artist in a playlist as a proxy for a "rating" of 1, indicating that the artist is relevant to the playlist.

The Co-Occurrence based algorithm, on the other hand, will measure the similarity between items based on how often they appear in the same playlist.

Preparing training Dataset

We'll start by creating a training dataset.

The first function reads every CSV file and combines them into a single pandas DataFrame. For artist similarity, we are only interested in the presence of artists in a playlist. Here's the code for reading the raw dataset:

def read_for_artists(allowlist):
    p = Path('/kaggle/input/spotify-playlists-csv')
    paths = list(p.iterdir())
    df = pd.concat((pd.read_csv(f, usecols=['pid', 'artist_uri']) for f in paths), ignore_index=True)
    df = df[df.artist_uri.isin(allowlist)]
    return df

As previously mentioned, I'll further limit the number of artists. By playing with the data, I noticed that it's reasonable to keep around 1K of them, corresponding to 10K playlist mentions. Here's the code for preparing the artist allowlist:

N_10k = 10_000 # leaves ~1K  artists
N_50k = 50_000 # leaves ~200 artists
N_88k = 88_000 # leaves ~100 artists

def get_common_artists_index(mentions):
    common_artists = pd.read_csv('/kaggle/input/common-spotify-artists/common_artist_uri.csv')
    df = common_artists[common_artists['count'] > mentions] 
    df = df.set_index('artist_uri')
    return df.index

To remove noise, I suggest to remove playlists that have too few or too many artists. Having too many artists might mean that this is a playlist with random songs, while having too few may risk losing signals and overfitting the model.

def clean_by_pid(df, p_min, p_max):    
    pid_counts = df.groupby('pid')['pid'].count()

    drop_df = ((pid_counts < p_min) | (pid_counts > p_max))
    drop_index = drop_df[drop_df].index
    df = df[~df.pid.isin(drop_index)]

    return df

Having more than one track from the same artist in a playlist might signal something, but I didn't want to introduce the complexity of a value function, so let's leave only unique mentions of artists in the training set.

def preprocess_artists(df):
    df.drop_duplicates(subset=['pid', 'artist_uri'], inplace=True)
    return df

After performing all necessary manipulations on the dataset, we are left with relevant (pid, artist) pairs.

The first two algorithms we will consider are variants of Collaborative Filtering. Collaborative Filtering relies on the notion of ratings, so for every pair of pid and artist, we’ll assign a rating of 1 to indicate that the artist is relevant to the playlist.

Additionally, it will be useful to index the dataset by the (pid, artist) pair. This will allow us to efficiently union dataset when we add negative data.

def prepare_positive(df, item_name):
    df['rating'] = 1
    df = df.set_index(['pid', item_name])
    return df

Now, let's move on to generating negative labels as promised. To do this, we will need to obtain the unique playlist IDs and unique artists in our dataset. We can then randomly match them a specified number of times to create our negative examples:

def generate_negative_pairs(N, unique_pids, unique_items):    
    items = np.random.choice(unique_items, size=N)
    pids = np.random.choice(unique_pids, size=N)

    df = pd.concat([pids, items], axis=1)
    df['rating'] = 0
    return df

def generate_negative_artists(N, pids, artists):
    df = generate_negative_pairs(N, pids, artists)
    df = df.set_index(['pid', 'artist_uri'])
    return df

To create a dataset, we can generate two random arrays: one for artists and one for playlists, this process can duplicate items if necessary. Then we can join the arrays into a single dataset by adding them as columns. It is important to note that we will need to index this new dataset accordingly.

To finalize the creation of our positive and negative datasets, we join them together and reset the indexed columns:

def join_dfs(df1, df2):
    res = pd.concat([df1, df2], sort=True, copy=False)
    # make sure to free the occupied memory
    df1.drop(df1.index, inplace=True)
    df2.drop(df2.index, inplace=True)
    return res


def reindex(df):
    df.reset_index(inplace=True)
    return df

Here is a summary of the steps we have discussed for creating a training dataset for collaborative filtering:

Obtain the list of the most popular artists to use as an allowlist.
Read in the raw data.
Clean up the noisy playlists.
Generate the negative dataset by randomly generating pairs of playlist IDs and artists that do not already exist in the dataset.
Assign positive and negative labels to the data accordingly.
Join datasets together into a single dataset.

Here is a function that combines these steps into a single flow for creating the training dataset for collaborative filtering:

def artist_training_set(n_common, pid_min_max, negative_ratio):
    # read most popular artists
    allowed_artists = get_common_artists_index(n_common)
    # read and preprocess positive labels data
    df = read_for_artists(allowed_artists)
    df = clean_by_pid(df, pid_min_max[0], pid_min_max[1])
    df = preprocess_artists(df)
    # generate negative labels
    df2 = generate_negative_artists(math.ceil(negative_ratio * len(df)), df.pid.drop_duplicates(), allowed_artists.to_series())
    # prepare negative labels
    df = prepare_positive(df, 'artist_uri')
    # finilise resulting dataset
    df = join_dfs(df, df2)
    return reindex(df)

If we execute this function, the resulting data will look this way:

training set

training_df = artist_training_set(n_common=N_10k,                                                     
                                  users_min_max=(5, 40),                                               
                                  negative_ratio=1,)

pid     artist_uri              rating
896625  7DMveApC7UnC2NPfPvlHSU  1
896625  23fqKkggKUBHNkbKtXEls4  1
896625  45eNHdiiabvmbp4erw26rg  1
896625  0L8ExT028jH3ddEcZwqJJ5  1
896625  2wY79sveU1sp5g7SokKOiI  1
... ... ... ...
957041  70BYFdaZbEKbeauJ670ysI  0
587936  4Z8W4fKeB5YxbusRsdQVPb  0
656290  0uq5PttqEjj3IH1bzwcrXF  0
   757  2RhgnQNC74QoBlaUvT4MEe  0
677025  2NhdGz9EDv2FeUw6udu2g1  0

Training Models

Now that the training dataset is prepared, we can experiment with various item similarity algorithms. As I mentioned, we will begin with Collaborative Filtering, which has two variations: Matrix Factorization and Collaborative Filtering with Multi-Layer Perceptron (MLP). Both methods follow a similar process: we create a vector representation for each artist, treat this vector as a location in N-dimensional space, and find the K closest artists to a given artist. The main difference lies in how these vectors are computed.

I have tried to make this article as practical as possible. If you are interested in learning more about the underlying theory behind these algorithms, I recommend the following resources:

The Netflix Tech Blog has an in-depth article on the collaborative filtering algorithms used by Netflix.
This tutorial from the DataCamp blog provides a practical introduction to implementing collaborative filtering in Python:
This course from Coursera, "Recommendation Systems and Deep Learning in Python," covers collaborative filtering in detail and includes hands-on exercises:
Article with theory behind how Collaborative Filtering algorithms work.

We will be utilizing the fastai library, so let's go ahead and create a Dataset Loader that retrieves data from the created DataFrame:

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
dls = CollabDataLoaders.from_df(df, item_name='artist_uri', rating_name='rating', username='pid', bs=4096, device=device)

Internally, the Dataset Loader will load the data into a device, encode URIs, shuffle and batch the data, and create a validation subset. Let's make sure it has got the correct data:

data batch

dls.show_batch()
pid     artist_uri              rating
902723  61lyPtntblHJvA7FMMhi7E  1
28782   7iZtZyCzp3LItcw1wtPI3D  1
41514   11Y54BxlxC3UIAUkU2eadQ  0
682607  4EVpmkEwrLYEg6jIsiPMIb  0
166133  3iri9nBFs9e4wN7PLIetAw  0
886234  7jefIIksOi1EazgRTfW2Pk  1
688805  1f5GqyOPo0CkotzzRwviBu  1
645015  1yAwtBaoHLEDWAnWR87hBT  1
14524   0n94vC3S9c3mb2HyNAOcjg  0
909486  7H55rcKCfwqkyDFH9wpKM6  1

Matrix Factorization Collaborative Filtering

I would like to begin by using the Matrix Factorization Collaborative Filtering item similarity method. Let's create and train the model:

learn = collab_learner(dls, n_factors=50, y_range=(0, 1))
epochs = 5
learning_rate = 5e-3
weight_decay = 0.1
learn.fit_one_cycle(epochs, learning_rate, wd=weight_decay)

Here are model metrics:

epoch	train_loss	valid_loss	time
0	0.217431	0.217432	01:46
1	0.216207	0.216138	01:47
2	0.193768	0.194263	01:48
3	0.171074	0.176515	01:47
4	0.160071	0.174029	01:49

One of the most important parameters when creating the model is the n_factors argument. This parameter specifies the size of the N-dimensional vector. A larger value allows the model to capture more concepts, but it will also require more data to be trained effectively.

After completing the model fitting process, we can check the embeddings of the items (artists):

learn.weight(['61lyPtntblHJvA7FMMhi7E'], is_item=True)

tensor([[ 0.1625, -0.3502, -0.5267, -0.4166, -0.5810,  0.8513, -0.4677,  0.8700,
         -0.4015,  0.9194,  0.4940,  0.5556, -0.6228, -0.0939, -0.1931,  0.3866,
          0.5970, -0.7663,  0.7816,  0.3170, -0.8104,  0.1340, -0.5477,  0.9246,
         -0.2875, -0.0404,  0.0587,  0.8526, -0.3287, -0.6467,  0.2383,  0.0480,
          0.3321, -0.5684, -0.6459,  0.6053,  0.2226,  0.2793, -0.3024,  0.0718,
         -0.4428, -0.1309, -0.2367, -1.0821,  0.3814, -0.0182, -0.1722,  0.4477,
          0.1380,  0.3781]])

These vectors may not be interpretable to humans, but we can use them to find similar items by calculating the cosine (or some another) distance between two vectors.

Now, we can construct a similarity matrix using another popular algorithm: a KNN classifier. To do this, we will use the sklearn library.

The process is as follows: train the KNN model and then, for each artist in the dataset, retrieve the N most similar artists.

Here is how we train the model:

weights = learn.model.i_weight.weight.cpu().detach().numpy()
items = dls.classes['artist_uri'].items
knn = KNeighborsClassifier(n_neighbors=20, algorithm='kd_tree')
knn.fit(weights, items)

Here is how we can retrieve similar items:

item = learn.weight(['3TVXtAsR1Inumwj472S9r4'], is_item=True).cpu().numpy()
knn.kneighbors(item, return_distance=False)

array([[ 520, 1079,  973,  412,   91,  268, 1133,  952,   36,   50,  816,
        1137,  886,  670,  775,   48,  644,  317,  346,  148]])

As you can see, the returned items are not URIs but indices, so let's convert them to URIs using the DatasetLoader:

see similar items

[dls.classes['artist_uri'].items[i] for i in neighb[0,1:]]

['7CCjtD0hCK005Bvg2WG1a7',
 '6d47Z08T4snK50HgTEHo5Z',
 '2jku7tDXc6XoB6MO2hFuqg',
 '0cGUm45nv7Z6M6qdXYQGTX',
 '1tqhsYv8yBBdwANFNzHtcr',
 '7kFfY4UjNdNyaeUgLIEbIF',
 '6TIYQ3jFPwQSRmorSezPxX',
 '0HjYETXAvcL6mzaKjAmH2K',
 '0NWbwDZY1VkRqFafuQm6wk',
 '5bgfj5zUoWpyeVatGDjn6H',
 '7mDU6nMUJnOSY2Hkjz5oqM',
 '67nwj3Y5sZQLl72VNUHEYE',
 '4TsHKU8l8Wq7n7OPVikirn',
 '5G9kmDLg3OeUyj8KVBLzbu',
 '0NIIxcxNHmOoyBx03SfTCD',
 '4LLpKhyESsyAXpc4laK94U',
 '27mFvqQj8KpjmdKIqcw1mG',
 '2HPaUgqeutzr3jx5a9WyDV',
 '137W8MRPWKqSmrBGDBFSop']

As previously, you can follow https://open.spotify.com/artist/URI link to see artists,
i.e. https://open.spotify.com/artist/7CCjtD0hCK005Bvg2WG1a7

Multi-Layer Perceptron Collaborative Filtering

Now I would like to try a different algorithm for computing embeddings: the Multi-Layer Perceptron.

The data preparation step is the same, as well as instantiation of a DataLoader. The main difference lies in the model definition, with fastai we can do this:

learn = collab_learner(dls, use_nn=True, layers=[50, 50], y_range=(0, 1))

In this case, we will let the fastai library to decide on the size of the embeddings and only specify the NN layers.

Let's train the model using the same parameters:

learn.fit_one_cycle(5, 5e-3, wd=0.1)

Model metrics:

epoch	train_loss	valid_loss	time
0	0.167988	0.167022	07:35
1	0.142461	0.144335	07:45
2	0.127324	0.134424	07:27
3	0.099502	0.128019	07:14
4	0.056717	0.137510	07:13

Training and using the KNN model is slightly different due to internal structure difference in fastai embeddings model:

weights = learn.model.embeds[1].weight.cpu().detach().numpy()
items = dls.classes['artist_uri'].items
knn = KNeighborsClassifier(n_neighbors=20, algorithm='kd_tree')
knn.fit(weights, items)

And this is how we use it:

item_idx = dls.classes['artist_uri'].o2i['3TVXtAsR1Inumwj472S9r4']
item = learn.model.embeds[1].weight[[item_idx]].cpu().detach().numpy()
neighb = knn.kneighbors(item, return_distance=False)

array([[ 520,  220,  780,  846,   90, 1081,  387, 1054,   57,  151,  754,
         894, 1000,  104,  148, 1029, 1128,  199,  156,  300]])

converted artist URIs

['1Xyo4u8uXC1ZmMpatF05PJ',
 '5K4W6rqBFWDnAN6FQUkS6x',
 '5pKCCKE2ajJHZ9KAiaK11H',
 '0c173mlxpT3dSFRgMO8XPh',
 '7CajNmpbOovFoOoasH2HaY',
 '2YZyLoL8N0Wb9xBt1NhZWg',
 '73sIBHcqh3Z3NyqHKZ7FOL',
 '0QHgL1lAIqAw0HtD7YldmP',
 '13ubrt8QOOCPljQ2FL1Kca',
 '55Aa2cqylxrFIXC767Z865',
 '69GGBxA162lTqCwzJG5jLp',
 '6l3HvQ5sa6mXTsMTB19rO5',
 '0fA0VVWsXO9YnASrzqfmYu',
 '137W8MRPWKqSmrBGDBFSop',
 '6vWDO969PvNqNYHIOW5v0m',
 '7iZtZyCzp3LItcw1wtPI3D',
 '1RyvyyTE3xzB2ZywiAwp0i',
 '163tK9Wjr9P9DmM0AVK7lm',
 '246dkjvS1zLTtiykXe5h60']

Storing similarities

We are now able to query similar items for a given item, so the next step is to compute a similarity matrix and store it in a file which we will use in the second part of this tutorial.

I plan to use the following structure and write it in a plain text:

seed_item1 similar_11 similar_12 similar_13 ...
seed_item2 similar_21 similar_22 similar_23 ...
...
seed_itemN similar_N1 similar_N2 similar_N3 ...

You may find this structure too simple, but in the next part I'll explain why we need it this way.

The first step is to compute the similarity matrix. To do this, we will need a few helper functions, the most notable of them is get_similar, which queries similar items for a given batch of items. The other functions should be self-explanatory:

def get_emb(learn, items):
    if isinstance(learn.model, EmbeddingNN):
        emb = learn.model.embeds[1].weight[items]
    else:
        emb = learn.model.i_weight.weight[items]
    return emb.cpu().detach().numpy()

def get_similar(knn, learn, items):
    emb = get_emb(learn, items)
    return knn.kneighbors(emb, return_distance=False)

def to_uri(dls, i):
    return dls.classes['artist_uri'].items[i]

def total_items(dls):
    return len(dls.classes['artist_uri'].items)

We can now construct a generator that will compute the similarity matrix, but won't store it in memory. This generator operates in batches when querying the KNN model, but later flattens the batch and iterates over a single row. Querying the KNN model in batches, hopefully, increases the throughput as it can benefit from CPU caches and vector instructions.

def compute_similarity(learn, dls, knn, bs):
    items = total_items(dls)

    stream = (
        (idx, similar_arr)
        for start_idx in range(0, items, bs)
        for (idx, similar_arr) in zip(
            range(start_idx, start_idx+bs),
            get_similar(knn, learn, slice(start_idx, start_idx+bs)),
        )
    )

    return (
        (to_uri(dls, idx), [to_uri(dls, i) for i in similar if i != idx])
        for idx, similar in stream
    )

The final piece of the puzzle is a loop that goes through the generator and writes the rows to a file:

def store_similarity_pairs(stream, name):
    with open(name, 'w') as f:
        for k, v in stream:
            f.write(k)
            f.write(' ')
            f.write(' '.join(v))
            f.write('\n')

With that, we have completed the Collaborative Filtering part. Let's move on to another similarity algorithm.

Co-ocurrence Counting

The next similarity algorithm is called Co-ocurrence Counting. Its idea is very simple: it counts the co-occurrences of items within a given window of events. Co-occurrences that happen often (more than the specified threshold) can be considered similar. In the case of Spotify dataset, instead of windows, we can count the co-occurrences of artists within a playlist.

For this algorithm, we are only interested in pairs of playlist ID and artist, and we don't need negative data or ratings.

One good aspect is that we can reuse most of the data processing logic from the previous example. This is how we read and process the data:

allowed_artists = get_common_artists_index(N_10k)
df = read_for_artists(allowed_artists)
df = clean_by_pid(df, 5, 30)
df = preprocess_artists(df)

pid     artist_uri
896625  7DMveApC7UnC2NPfPvlHSU
896625  23fqKkggKUBHNkbKtXEls4
896625  45eNHdiiabvmbp4erw26rg
896625  0L8ExT028jH3ddEcZwqJJ5
896625  2wY79sveU1sp5g7SokKOiI
...     ...
59361   2Q0MyH5YMI5HPQjFjlq5g3
59361   3XHO7cRUPCLOr6jwp8vsx5
59361   42vg2T0Xg9yPaAgogJzoQH
59361   0YhUSm86okLWldQVwJkLlP
59364   3IYUhFvPQItj6xySrBmZkd

The number of unique artist pairs can increase by several orders of magnitude from the initial dataset size. In such case, using a pair of strings (string, string) is very inefficient and will have a large memory footprint. However, we can encode (factorize) the artist IDs into unique integers and store pairs of integers (int, int) instead. This will reduce the memory footprint by a factor of 10.

df['artist_idx'], i2a = df.artist_uri.factorize()

pid artist_uri  artist_idx
896625  7DMveApC7UnC2NPfPvlHSU  0
896625  23fqKkggKUBHNkbKtXEls4  1
896625  45eNHdiiabvmbp4erw26rg  2
896625  01aC2ikO4Xgb2LUpf9JfKp  3
896625  0L8ExT028jH3ddEcZwqJJ5  4
... ... ...
59361   42vg2T0Xg9yPaAgogJzoQH  1348
59361   0YhUSm86okLWldQVwJkLlP  1305
59361   42crL07E4WPfVovyUtMpvC  311
59364   2qFr8w5sWUITRlzZ9kZotF  1330
59364   3IYUhFvPQItj6xySrBmZkd  489

Now, we can generate all present combinations of artist pairs from every playlist:

def coco_pairs(df):
    groupped = df.groupby('pid')['artist_idx'].apply(list)
    return ((item1, item2) if item1 < item2 else (item2, item1)
            for _, arr in groupped.items()
            for item1, item2 in itertools.combinations(arr, r=2))

The logic is as follows: we group the dataset by playlist ID and aggregate the artists into a list. Then, for each list, generate combinations, ensuring that the pair is ordered so that (A, B) is the same as (B, A). This is because the order of the pair does not matter. I am using a generator because storing every pair before aggregating them would almost certainly not fit in memory. The output of this generator is a single artist co-occurrence pair from a playlist.

We can now use Python's Counter class to count the co-occurrences of pairs:

c = Counter(coco_pairs(df))

Let's check the most common co-occurrences:

c.most_common(3)

[((77, 434), 9175), ((30, 77), 8889), ((77, 97), 8474)]

And convert the encoded numerical IDs back to artist URIs using the previously saved index:

{i: i2a[i] for i in [77, 434, 30]}

{77: '3TVXtAsR1Inumwj472S9r4',
 434: '5K4W6rqBFWDnAN6FQUkS6x',
 30: '2YZyLoL8N0Wb9xBt1NhZWg'}

We will create a DataFrame with three columns: item1, item2, and count. Then, group the DataFrame by item1 and get the item2 values that are similar to it, so the item order now matters. Therefore, we will need to duplicate the counter and store both the (A, B) pair and the (B, A) pair:

o1 = ((i1, i2, c[i1, i2]) for (i1, i2) in c)
o2 = ((i2, i1, c[i1, i2]) for (i1, i2) in c)

coco = pd.DataFrame(itertools.chain(o1, o2), columns=['item1', 'item2', 'count'])

item1   item2   count
869     1623    8
535     869     8
536     869     10
224     869     131
869     1279    6
...     ...     ...
811     120     1
251     120     1
1119    120     1
1352    120     1
1582    1566    1

Let's apply a threshold and see what similarities the co-occurrence counting model yields:

similarities = res[res['count'] > 1000]

item1   item2   count
142     714     1085
236     280     1531
142     236     1046
280     471     1105
142     280     1349
...     ...     ...
450     252     1085
219     218     1379
450     198     1116
677     1       1162
677     259     1347

And convert encoded ints for the first result:

i2a[142], i2a[714]
('1dfeR4HaWDbWqFHLkxsg1d', '26bcq2nyj5GB7uRr558iQg')

This might be a final solution, but we're overlooking an important aspect when we filter pairs in this way: less popular artists will always have fewer co-occurrences, yet they may still be similar to each other. To address this, we can add a normalisation step.

First, let's count the total number of mentions of each artist:

item_counts = res.groupby('item1')['count'].sum()

item1
0        71840
1       193403
2        29229
3       106297
4       182767
         ...  
1168     18829
1169      9948
1170     24442
1171     12039
1172      8835

Then, we can add a normalised count to each pair, which is computed as pair count diveded by a sum of total counts of individual items:

pair_count = res['count']
item1_total = res.item1.map(item_counts)
item2_total = res.item2.map(item_counts)
res['normalised_count'] = pair_count / (item1_total + item2_total)

This normalised count penalises popular artists and boosts less popular.

item1   item2   count   normalised_count
320     714     10      0.000180
236     714     178     0.001518
280     714     202     0.001310
471     714     348     0.003135
714     1071    224     0.003058
...     ...     ...     ...
1088    616     1       0.000015
616     492     1       0.000016
616     517     1       0.000021
710     660     1       0.000030
591     219     1       0.000012

If we apply the threshold to the normalised count, we can see how different pairs that were previously missing or present have been affected:

similarities = res[res.normalised_count > 0.004]

item1   item2   count   normalised_count
472     714     312     0.004030
142     714     1085    0.005807
647     714     605     0.005391
236     280     1531    0.008224
236     471     915     0.006402
...     ...     ...     ...
1172    1169    111     0.005910
1169    1166    89      0.005227
1135    1093    206     0.010754
1137    744     178     0.004037
1169    696     78      0.004469

For example, the last pair (1169, 696) which has only 78 co-occurrences would have certainly been filtered out previously. These artist URIs are shown below, you can judge for yourself how similar they are:

i2a[1169], i2a[696]
('4sTFGCigAQIUiEy8wSSQNF', '1YSA4byX5AL1zoTsSTlB03')

Storing similarities

Let's now create the similarity matrix and write it to a file in the same format as before:

similar_items = similarities.groupby('item1').item2.apply(list)

item1
0                                                                     [1, 408]
1                [200, 252, 259, 57, 202, 408, 896, 66, 467, 285, 334, 677, 0]
2                                               [333, 284, 331, 858, 330, 674]
3                                               [234, 276, 648, 767, 238, 631]
4                [126, 204, 228, 998, 506, 167, 131, 326, 1122, 103, 192, 288]
                                         ...                                  
1169    [1172, 1167, 862, 863, 864, 1165, 779, 726, 731, 636, 1059, 1166, 696]
1170                                      [608, 145, 1038, 473, 989, 147, 387]
1171                [700, 746, 1074, 744, 602, 1154, 114, 745, 1130, 747, 611]
1172                                        [696, 1091, 1166, 744, 1165, 1169]

One downside of this algorithm is that it doesn't provide a guaranteed number of recommendations. For example, item {idx=0} has only 2 similar items, while item {idx=1169} has 13, etc.

And finally we can write them to the text file:

fname = 'coco.txt'
with open(fname, 'w') as f:
    for k, v in similar.items():
        f.write(i2a[k])
        f.write(' ')
        v = [i2a[i] for i in v[:20]]
        f.write(' '.join(v))
        f.write('\n')

Since this article is getting quite long, I'll leave it as an exercise for you to model the track similarities.

This concludes the section on finding similar items. We overviewed how to build similarity matrices using Collaborative Filtering and Co-occurence counting algorithms.
You can find an example matrix files in this GitHub Gist.

You can also try other algorithms that produce embeddings, such as Word2Vec, or even possibly content-based algorithms that take artist or track features into account.
Share your artist or track similarity results in the comments!

In the next chapter, I'll cover how we can use similarities to retrieve recommendations. If you don't want to miss it, be sure to hit the follow button.

Thank you for reading this far. I hope it was interesting!

How I improved my JS code with these 3 functions

Vladyslav Hutov — Mon, 28 Nov 2022 16:05:43 +0000

I’m a huge fan of Functional Programming. Today I will show you how it helps writing readable code.

I worked recently on a recommendation system in Node.js, where I applied some of FP principles.

Initial code

In my examples, I’ll use the Flow type system so that it’s easier to follow the idea. The code consists of two parts: a service API and a recommendation flow.

Service API

First, here are the type definitions we'll use in the service.

type Options = { [string]: any }

type Entity = { [string]: any }

type OneOrMany<A> = A | A[]

And this is the service interface:

class RecommendationService {
    user =           async (input: OneOrMany<number>)                          : Promise<Entity[]> => []
    signals =        async (input: OneOrMany<Entity>, options: Options)        : Promise<Entity[]> => []
    signalCreators = async (input: OneOrMany<Entity>, options: Options)        : Promise<Entity[]> => []
    creatorContent = async (input: OneOrMany<Entity>, options: Options)        : Promise<Entity[]> => []
    similar =        async (input: OneOrMany<Entity>, options: Options)        : Promise<Entity[]> => []
    enrich =         async (input: OneOrMany<Entity>)                          : Promise<Entity[]> => []
    set =            async (input: OneOrMany<Entity>, from: string, to: string): Promise<Entity[]> => []
    take =           async (input: OneOrMany<Entity>, amount: number)          : Promise<Entity[]> => []
    shuffle =        async (input: OneOrMany<Entity>)                          : Promise<Entity[]> => []
    deduplicate =    async (input: OneOrMany<Entity>, by: string)              : Promise<Entity[]> => []
    sort =           async (input: OneOrMany<Entity>, by: string)              : Promise<Entity[]> => []
    merge =          async (...inputs: Promise<Entity[]>[])                    : Promise<Entity[]> => []
}

We will concentrate on how we compose those functions, so their implementation doesn’t matter to us.

Most have a signature of expecting an input with optional configurations and returning an array of modified data. This API is very flexible and allows the designing of a variety of recommendation flows.

Recommendation flow

I wanted to use this API in the following way:

At a high level, it has two parallel recommendations: via similar content from user signals and via similar creators of those signals, with many additional steps on top of those.

Let see how we can describe this diagram. We will start with chaining promises like this:

recs.user(userId)
    .then((i) => recs.signals(i, {...}))

And merging them like this:

recs.merge(
    flow1,
    flow2
)

Also, merge returns promise as well, so we can chain it:

recs.merge(...)
    .then((i) => g(i))

The full flow implementation would look this way:

async function recFlow(userId: number, recs: RecommendationService): Promise<Entity[]> {
    const userFlow: Promise<Entity[]> = recs
        .merge(
            recs.user(userId).then((i) => recs.signals(i, { someConfig: 'A' })),
            recs.user(userId).then((i) => recs.signals(i, { someConfig: 'B' }))
        )
        .then((i) => recs.deduplicate(i, 'id'))
        .then((i) => recs.set(i, 'id', 'origin'))
        .then((i) => recs.similar(i, { someOption: 1 }))
        .then((i) => recs.deduplicate(i, 'id'))
        .then((i) => recs.shuffle(i))
        .then((i) => recs.take(i, 10))

    const creatorFlow: Promise<Entity[]> = recs
        .merge(
            recs.user(userId).then((i) => recs.signalCreators(i, { someConfig: 'A' })),
            recs.user(userId).then((i) => recs.signalCreators(i, { someConfig: 'B' }))
        )
        .then((i) => recs.deduplicate(i, 'id'))
        .then((i) => recs.set(i, 'id', 'origin'))
        .then((i) => recs.similar(i, { someOption: 1 }))
        .then((i) => recs.deduplicate(i, 'id'))
        .then((i) => recs.creatorContent(i, {}))
        .then((i) => recs.shuffle(i))
        .then((i) => recs.take(i, 10))

    const bothFlows: Promise<Entity[]> = recs
        .merge(userFlow, creatorFlow)
        .then((i) => recs.deduplicate(i, 'id'))
        .then((i) => recs.enrich(i))
        .then((i) => recs.sort(i, 'order'))

    return await bothFlows
}

The first question you could have is: “why don’t we use async/await?”.

If we want to use the await keyword to chain computations, we’ll need to split this code into multiple smaller sub-functions because the merge function expects parallel promises. In this case, it’s undesirable as we want to keep the complete picture of the entire flow, hence we chain promises with then.

To me, this code is very cumbersome and cluttered. We concentrate too much on chaining functions rather than business logic.

Additionally, I don’t like this indirection in the merge: when you see this method, you need to put the reading context into a stack and explore what happens inside the function arguments.

Now, let’s improve this code, but first, I’ll introduce you to currying:

Currying is a technique of translating a function with multiple arguments into a sequence of functions with a single argument:

\Rightarrow F(a)(b)(c)

It allows the creation of partially applied functions, i.e. capturing intermediate arguments and constructing new functions:

\\ x = H(3) \\ y = H(4)

Refactoring service

Let’s refactor the RecommendationsService using the currying technique. For this, I want to introduce a new type alias:

type Pipe<A> = (OneOrMany<A> => Promise<Entity[]>)

Pipe is a generic async function which receives a single argument 'input' and returns an 'output'.

Now, let’s see how we can rewrite the service using Pipe and currying:

class RecommendationService {
    user                                       : Pipe<number> =  async (input) => []
    signals                = (options: Options): Pipe<Entity> => async (input) => []
    signalCreators         = (options: Options): Pipe<Entity> => async (input) => []
    creatorContent         = (options: Options): Pipe<Entity> => async (input) => []
    similar                = (options: Options): Pipe<Entity> => async (input) => []
    enrich                                     : Pipe<Entity> =  async (input) => []
    set            = (from: string, to: string): Pipe<Entity> => async (input) => []
    take                     = (amount: number): Pipe<Entity> => async (input) => []
    shuffle                                    : Pipe<Entity> =  async (input) => []
    deduplicate                  = (by: string): Pipe<Entity> => async (input) => []
    sort                         = (by: string): Pipe<Entity> => async (input) => []
    merge = async (...inputs: Promise<Entity[]>[]): Promise<Entity[]> => []
}

It may not be clear at first where the currying is, so let me show some methods without types:

class RecommendationService {
    user = async(input) => []
    ...
    signals = (options) => async (input) => []
    ...
    set = (from, to) => async (input) => []
    ...
}

This way, we transformed signals from (input, options) to a curried version of (options)(input).

Some functions, like user or shuffle, don’t expect any configuration and are instances of Pipe type already; others, like signals or similar, receive some options and create the Pipe instance.

With this new API, we can now change the flow implementation:

async function recFlow(userId: number, recs: RecommendationService): Promise<Entity[]> {
    const userFlow: Promise<Entity[]> = recs
        .merge(
            recs.user(userId).then(recs.signals({ someConfig: 'A' })),
            recs.user(userId).then(recs.signals({ someConfig: 'B' }))
        )
        .then(recs.deduplicate('id'))
        .then(recs.set('id', 'origin'))
        .then(recs.similar({ someOption: 1 }))
        .then(recs.deduplicate('id'))
        .then(recs.shuffle)
        .then(recs.take(10))

    const creatorFlow: Promise<Entity[]> = recs
        .merge(
            recs.user(userId).then(recs.signalCreators({ someConfig: 'A' })),
            recs.user(userId).then(recs.signalCreators({ someConfig: 'B' }))
        )
        .then(recs.deduplicate('id'))
        .then(recs.set('id', 'origin'))
        .then(recs.similar({ someOption: 1 }))
        .then(recs.deduplicate('id'))
        .then(recs.creatorContent({}))
        .then(recs.shuffle)
        .then(recs.take(10))

    const bothFlows: Promise<Entity[]> = recs
        .merge(userFlow, creatorFlow)
        .then(recs.deduplicate('id'))
        .then(recs.enrich)
        .then(recs.sort('order'))

    return await bothFlows
}

It looks much better already, as we eliminated some clutter from passing the input everywhere. At this stage, I’m more or less satisfied with the API. However, there is still a lot to improve in the flow, and we haven't fixed the merge flaw yet.

Before moving to the next step, I need to introduce Ramda - functional library for JavaScript. It comes with plenty of useful utility functions.
$ npm install ramda
const R = require('ramda')

Chaining

Let’s meet the first two functions which help make this code better: R.pipeWith and R.andThen.

Together, they can compose multiple Pipes into an ultimate Pipe. For example, the following two functions are identical:

const userSignalsVanilla: Pipe<number> = 
  async (userId) => recs.user(userId).then(recs.signals({}))

const userSignalsRamda: Pipe<number> = 
  R.pipeWith(R.andThen, [recs.user, recs.signals({})]

Now, we can create a helper function flow, which combines a list of Pipes.

const flow = <Inp>(f: Pipe<Inp>, ...fs: Pipe<Entity>[]): Pipe<Inp> => 
  R.pipeWith(R.andThen, [f].concat(fs))

Please, notice that the resulting type of Pipe, i.e. input type, is the same as the type of the first function in the chain.

Having this utility function, we can rewrite our flow in this way:

async function recFlow(userId: number, recs: RecommendationService) {

    const userFlow: Pipe<number> = flow(
        (i) =>
            recs.merge(
                flow(recs.user, recs.signals({ someConfig: 'A' }))(i),
                flow(recs.user, recs.signals({ someConfig: 'B' }))(i)
            ),
        recs.deduplicate('id'),
        recs.set('id', 'origin'),
        recs.similar({ someOption: 1 }),
        recs.deduplicate('id'),
        recs.shuffle,
        recs.take(10)
    )

    const creatorFlow: Pipe<number> = flow(
        (i) =>
            recs.merge(
                flow(recs.user, recs.signalCreators({ someConfig: 'A' }))(i),
                flow(recs.user, recs.signalCreators({ someConfig: 'B' }))(i)
            ),
        recs.deduplicate('id'),
        recs.set('id', 'origin'),
        recs.similar({ someOption: 1 }),
        recs.deduplicate('id'),
        recs.creatorContent({}),
        recs.shuffle,
        recs.take(10)
    )

    const bothFlows: Pipe<number> = flow(
        (i) => recs.merge(userFlow(i), creatorFlow(i)),
        recs.deduplicate('id'),
        recs.enrich,
        recs.sort('order')
    )

    return await bothFlows(userId)
}

Let’s now fix the merge part.

Converging

Judging by its signature, the merge function takes multiple branches and merges their output into a single array. Ramda has a helper for this case either: R.converge, so again these two functions are equal:

const pipeVanilla: Pipe<number> = 
  async (i) => recs.merge(userFlow(i), creatorFlow(i))

const pipeRamda: Pipe<number> =
  R.converge(recs.merge, [userFlow, creatorFlow])

Final version of the flow:

async function recFlow(userId: number, recs: RecommendationService): Promise<Entity[]> {

    const userFlow: Pipe<number> = flow(
        recs.user,
        R.converge(recs.merge, [
            recs.signals({ someConfig: 'A' }), 
            recs.signals({ someConfig: 'B' })
        ]),
        recs.deduplicate('id'),
        recs.set('id', 'origin'),
        recs.similar({ someOption: 1 }),
        recs.deduplicate('id'),
        recs.shuffle,
        recs.take(10)
    )

    const creatorFlow: Pipe<number> = flow(
        recs.user,
        R.converge(recs.merge, [
            recs.signalCreators({ someConfig: 'A' }), 
            recs.signalCreators({ someConfig: 'B' })
        ]),
        recs.deduplicate('id'),
        recs.set('id', 'origin'),
        recs.similar({ someOption: 1 }),
        recs.deduplicate('id'),
        recs.creatorContent({}),
        recs.shuffle,
        recs.take(10)
    )

    const bothFlows: Pipe<number> = flow(
        R.converge(recs.merge, [
            userFlow, 
            creatorFlow
        ]),
        recs.deduplicate('id'),
        recs.enrich,
        recs.sort('order')
    )

    return await bothFlows(userId)
}

With the reduced clutter, now we can concentrate only on the business logic, instead of remembering to pass inputs and chain promises.

I encourage you to check out Ramda documentation for other useful utility functions, experiment with them and improve the readability of your code.

I hope you will find this article useful, thank you!

Latency: 6 questions you were afraid to ask

Vladyslav Hutov — Sat, 19 Nov 2022 21:33:47 +0000

Throughout my career, I’ve been collecting knowledge about performance piece by piece. It seemed to me as some kind of magic available to the chosen. In this post, I want to share with you my findings about such a topic as Application Latency.

Here I’ve collected 6 the most common questions you might have when struggling with application optimisations.

1. What is latency?

There is a lot of confusion in terms when talking about software engineering. This one is no exception. Ask ten engineers, and you will hear ten different answers.

In this article, I’m going to talk about a term that “Uncle Bob” Martin Fowler defined as “response time”:

Response time is the amount of time it takes for the system to process a request from the outside. This may be … a server API call. -- Martin Fowler

That is the time it takes from when a server receives a request to the time for it to respond.

Also, some may refer to latency as the total time it takes between a client making a request and receiving a response:

2Network\ latency\ 1 + Response\ time + Network\ latency\ 2

2. Should I care about latency?

Yes! If you want to create high-quality software, you absolutely should care about it. Users are affected by slow applications. According to this study, 1000 ms is enough to lose focus. But let me be clear, “care” ≠ “optimise”.

Premature optimization is the root of all evil -- Donald Knuth

Before jumping in and hunting for optimisations, you must understand that a good application is an equilibrium of characteristics: functional and non-functional. Foremost, it must satisfy user needs.

Ensure that you measure the app correctly. You can decide to optimise the app only when you have a good picture of its performance. Optimising is hard, and it doesn’t always make sense to spend weeks rewriting logic and going from 500ms to 400ms.

3. How to measure latency?

Usually, web servers have built-in tools for monitoring your app performance. They show only high-level metrics such as route response time. Having them is enough to judge if you need to work on optimisations but not enough to understand where to begin.

Tracing libraries can help you with both. They allow pinpointing where the poor performance occurs. An example of such a library is statstd. It provides a convenient API to monitor functions and code blocks. Check an example of this far-fetched Python app:

import statsd
import time

client = statsd.StatsClient()


@app.route("/")
@client.timer("HomeService.home")
def home():
    with client.timer("HomeService.database_calls"):
        message = _get_message()
        name = _get_name()
    page = _render_page(message, name)
    return page


@client.timer("HomeService._render_page")
def _render_page(message: str, name: str) -> str:
    time.sleep(0.1)
    return f"<p>{message} {name}</p>"


@client.timer("HomeService._get_message")
def _get_message() -> str:
    time.sleep(0.4)
    return "Hello"


@client.timer("HomeService._get_name")
def _get_name() -> str:
    time.sleep(0.5)
    return "world"

You can also rate limit it, e.g. log every 100th request, if your RPS is high enough.

When reading metrics, you need to decide which aggregation function to choose. Often the choice falls on “avg”. But measuring latency with avg is misleading.

Imagine that users open the home page with the following latencies: 100ms, 97ms, 312ms, 750ms, 60ms, 150ms, 809ms, and 121ms. The average latency is 300ms, which doesn’t describe the reality. For a better understanding, you can use percentiles (P50, P75, P95, P99, etc.).

For given example, percentiles are:

Percentile	Latency, ms
P50	136
P75	422
P95	788
P99	805

P95 of 788ms means that 95% of requests execute faster than 788ms. So when optimising your app, you mainly target 5% of users who wait more than 788ms for it to complete the request.

4. What causes latency?

You won’t be surprised if I tell you that many factors come into play. The list I provide is not a complete picture, but a good start.

Underprovisioned resource - CPU/Memory/Network doesn’t meet your workload requirements. It means that a request is awaiting resources.
I/O operations - calling external services, writing to files, logging, etc. are slow and cause your request to wait.
Other applications which run on the host are utilising its resources.
Inefficient code - when your application has complex business logic, ensure that you use optimal algorithms to solve the problem. It may also be related to the inefficient concurrent code (i.e. frequent context switches) or instantiation of heavy objects.
Garbage collection - some language environments like JVM come with a garbage collector making our life much easier. However, it comes at its cost, and GC can pause the entire app/thread execution and do a memory cleanup, which results in additional wait time for the request.
Different zones can cause network latency between your client and the app, e.g. a user from Europe calls a service in the US.

5. Does programming language matter?

It only matters in rare use cases. Modern languages have plenty of optimisations like JIT, AOT compilation, etc., making them run almost as fast as machine code.

6. How to improve latency?

Analyse DB access patterns, add indices and optimise queries. But remember, adding an index is a trade-off between read and write speed.
Cache data. Use local and/or remote in-memory caches, as this reduces the load from other external components like the database and allows it to handle responses much faster.
Defer work. Instead of making users wait, consider executing logic asynchronously, for example, passing messages to a durable queue and processing them later.
Precompute. Think if you can efficiently precompute and store computed data. For example, generate analytics reports on a scheduled basis instead of ad-hoc, and your users will be surprised by the response time.
Prefetch. Actually, it's not a latency improvement but a hack to "hide" it. Instead of waiting for the user's action, prefetch the most possible results. For example, if you have a search bar with suggestions, you can predict the subsequent letters and prefetch results, so that clients don't need to request them.
Scale. Improve resource utilisation by scaling instances of your app, database or other systems.
Use the right tools. Analyse the effectiveness of the existing app architecture and ensure that you use suitable tools for the problem, e.g. columnar vs row databases, pull vs push queues, binary vs text protocols, stream vs batch processing etc.
Monitor performance regressions as part of the CI/CD pipeline or set up alerts on latency metrics.

Let me know in the comments, if you have more questions.