Ingesting From Multiple Data Sources into Quine Streaming Graphs
Tags: quine, streaminggraph, datamodel
Published: 2022-06-06 18:12:57

---
title: Ingesting From Multiple Data Sources into Quine Streaming Graphs
published: true
description: Ingest from multiple data sources to create a single streaming graph. Covers CSV import.
tags: quine, streaminggraph, datamodel
cover_image: https://dev-to-uploads.s3.amazonaws.com/uploads/articles/5em1apiq0rpqr7tbcoqj.png
canonical_url: https://www.thatdot.com/blog/ingesting-from-multiple-data-sources-into-quine-streaming-graph
---

As part of the ongoing series in which I exploring different ways to use the ingest stream to load data into Quine, I want to cover one of Quine's specialities: building a streaming graph from multiple data sources. This time, we'll work with `CSV` data exported from IMDb to answer the question; *"Which actors have acted in and directed the same movie?"*

## The `CSV` Files
Usually, if someone says that they have data, most likely it's going to be in `CSV` format or pretty darn close to it. (Or `JSON`, but that is another blog post.) In our case, we have two files filled with data in `CSV` format. Let's inspect what's inside.

**File 1: `movieData.csv`**

The `movieData.csv` file contains records for actors, movies, and the actor's relationship to the movie. Conveniently, each record type has a schema, flattened into rows during export.  

Should we separate the data back into discrete files and then load them? No, we can set up separate ingest streams to act on each data type in the file. Effectively, we will separate the "jobs to do" into Cypher queries and stream in the data. 

**File 2: `ratingData.csv`**

Our second file, `ratingData.csv` is very straightforward. It contains 100,000 rows of movie ratings. Adding the `ratings` data into our model completes our discovery phase for the supplied data. 

![Original implied schema of IMDB data.](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/vvqftrx67iebnb3kcqdb.png)
 

## The `CypherCsv` Ingest Stream

The Quine API documentation defines the [schema](https://docs.quine.io/reference/rest-api.html#/schemas/com.thatdot.quine.routes.IngestStreamConfiguration) of the *File Ingest Format* ingest stream for us. The schema is robust and accommodates `CSV`, `JSON`, and `line` file types. Please take a moment to read through the documentation. Be sure to select `type: FileIngest` -> `format: CypherCsv` using the API documentation dropdowns. 

I define ingest streams to transform and load the movie data into Quine. Quine ingest streams behave independently and in parallel when processing files. This means that we can have multiple ingest streams operating on a single file. This is the case for the `movieData.csv` file because there are several operations that we need to perform on multiple types of data.

## Movie Rows

The first ingest stream that I set up will address the `Movie` rows in the `movieData.csv` file. There are 9125 movies in the data set. I create two nodes from each Movie row using an ingest query, `movie` and `genre`. I store all of the movie data as properties in the Movie mode. 

```cypher
WITH $that AS row
MATCH (m) WHERE row.Entity = 'Movie' AND id(m) = idFrom("Movie", row.movieId)
SET
  m:Movie,
  m.tmdbId = row.tmdbId,
  m.imdbId = row.imdbId,
  m.imdbRating = toFloat(row.imdbRating),
  m.released = row.released,
  m.title = row.title,
  m.year = toInteger(row.year),
  m.poster = row.poster,
  m.runtime = toInteger(row.runtime),
  m.countries = split(coalesce(row.countries,""), "|"),
  m.imdbVotes = toInteger(row.imdbVotes),
  m.revenue = toInteger(row.revenue),
  m.plot = row.plot,
  m.url = row.url,
  m.budget = toInteger(row.budget),
  m.languages = split(coalesce(row.languages,""), "|"),
  m.movieId = row.movieId
WITH m,split(coalesce(row.genres,""), "|") AS genres
UNWIND genres AS genre
WITH m, genre
MATCH (g) WHERE id(g) = idFrom("Genre", genre)
SET g.genre = genre, g:Genre
MERGE (m:Movie)-[:IN_GENRE]->(g:Genre)
```

Quine passes each line to the ingest stream via the variable `$that` to which I assign the identity `row`. A `MATCH` is made when the `row.Entity` value is `Movie` and a node `id` is returned from the `idFrom()` function. `SET` is used to give the node a label and to store metadata as node properties. 

Each movie row has a pipe `|` delimited list of genres in the `genres` column. I split the column value apart and created a Genre node for each genre in the list, labeled and containing the genre as a property. 

Finally, the `Movie` node is related to the `Genre` node with `MERGE`. 

## Person Rows
The second ingest stream addresses the `Person` rows in the same way I did for the `Movie` rows. There are 19047 person records in the `movieData.csv` file. 

```cypher
WITH $that AS row
MATCH (p) WHERE row.Entity = "Person" AND id(p) = idFrom("Person", row.tmdbId)
SET
  p:Person,
  p.imdbId = row.imdbId,
  p.bornIn = row.bornIn,
  p.name = row.name,
  p.bio = row.bio,
  p.poster = row.poster,
  p.url = row.url,
  p.born = row.born,
  p.died = row.died,
  p.tmdbId = row.tmdbId,
  p.born = CASE row.born WHEN "" THEN null ELSE datetime(row.born + "T00:00:00Z") END,
  p.died = CASE row.died WHEN "" THEN null ELSE datetime(row.died + "T00:00:00Z") END
```

The ingest query in this ingest stream matches when the `row.Entity` is `Person`, creates a node using the `idFrom()` function, and stores the Person metadata in node parameters. 

## Join Rows

Looking at the rows that have `Join` in the `Entity` column leads me to believe that the data in this `CSV` file originated from a relational database. There are two types of joins in the file, `Acted` and `Directed`. The ingest queries below process them. 

**Acted In**
```cypher
WITH $that AS row
WITH row WHERE row.Entity = "Join" AND row.Work = "Acting"
MATCH (p) WHERE id(p) = idFrom("Person", row.tmdbId)
MATCH (m) WHERE id(m) = idFrom("Movie", row.movieId)
MATCH (r) WHERE id(r) = idFrom("Role", row.tmdbId, row.movieId, row.role)
SET 
  r.role = row.role, 
  r.movie = row.movieId, 
  r.tmdbId = row.tmdbId, 
  r:Role
MERGE (p:Person)-[:PLAYED]->(r:Role)<-[:HAS_ROLE]-(m:Movie)
MERGE (p:Person)-[:ACTED_IN]->(m:Movie)
```

Acted join rows create relationships between Person, Role, and Movie nodes. There are two paths created from the Person nodes. The first path `(p)-[:PLAYED]->(r)<-[:HAS_ROLE]-(m)` establishes the relationship between actors (Person) and the roles they have played as well as the roles in a movie (Movies). A second path is formed that directly relates an actor to movies they acted in. 

**Directed**
```cypher
WITH $that AS row
WITH row WHERE row.Entity = "Join" AND row.Work = "Directing"
MATCH (p) WHERE id(p) = idFrom("Person", row.tmdbId)
MATCH (m) WHERE id(m) = idFrom("Movie", row.movieId)
MERGE (p:Person)-[:DIRECTED]->(m:Movie)
```

The Directed ingest query matches join rows and creates a path relating directors with the movies they have directed. 

## Ratings

```cypher
WITH $that AS row
MATCH (m) WHERE id(m) = idFrom("Movie", row.movieId)
MATCH (u) WHERE id(u) = idFrom("User", row.userId)
MATCH (rtg) WHERE id(rtg) = idFrom("Rating", row.movieId, row.userId, row.rating)
SET u.name = row.name, u:User
SET rtg.rating = row.rating,
  rtg.timestamp = toInteger(row.timestamp),
  rtg:Rating
MERGE (u:User)-[:SUBMITTED]->(rtg:Rating)<-[:HAS_RATING]-(m:Movie)
MERGE (u:User)-[:RATED]->(m:Movie)
```

The last ingest query processes rows from the `ratingData.csv` file. The query creates User and Rating nodes, then relates them together. 

## Running the Recipe
As my project progressed, I developed a [Quine recipe] (https://docs.quine.io//core-concepts/about-recipes.html#what-is-a-recipe) to load my `CSV` files and perform the analysis. Running the recipe requires a couple of Quine options to pass in the locations of the `CSV` files and an updated configuration setting. 

```shell
java \
-Dquine.in-memory-soft-node-limit=30000 \
-jar ../releases/latest -r movieData \
--recipe-value movie_file=movieData.csv \
--recipe-value rating_file=ratingData.csv
```

After ingesting the `CSV` files, it results in the data set stored in Quine:

![The data model in Quine for the IMDB data.](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/f0xn25prwet0rnlybht9.png)
 

The orange Movie and Person nodes are created directly from the `Entity` column in `movieData.csv`. The User node is from `ratingData.csv` and the green nodes were derived from data stored within an entity row. The `ActedDirected` relationship is built by the standing query in the recipe. 

## Answering the Question
Getting all of this data into Quine was only part of the challenge. Remember the question that we were asked, *"which actors have acted in and directed the same movie?"*

Quine is a streaming graph; if we were to connect the ingest streams to the streaming source, rather than `CSV` files, the [standing query](https://docs.quine.io/components/writing-standing-queries.html) inside of the recipe that I developed would answer the question for movies in the past as well as movies in the future.  

Our standing query matches when a complete pattern for the situation when an actor (`Person`) both `ACTED_IN` and `DIRECTED` the same movie. 

```cypher
MATCH (a:Movie)<-[:ACTED_IN]-(p:Person)-[:DIRECTED]->(m:Movie) 
WHERE id(a) = id(m)
RETURN id(m) as movieId, m.title as Movie, id(p) as personId, p.name as Actor
```

When the standing query completes a match, it processes the movie `id` and person `id`  through the [output](https://docs.quine.io/components/standing-query-outputs.html) query and actions.

```yaml
standingQueries:
  - pattern:
      type: Cypher
      mode: MultipleValues
      query: |-
        MATCH (a:Movie)<-[:ACTED_IN]-(p:Person)-[:DIRECTED]->(m:Movie) 
        WHERE id(a) = id(m)
        RETURN id(m) as movieId, m.title as Movie, id(p) as personId, p.name as Actor
    outputs:
      set-ActedDirected:
        type: CypherQuery
        query: |-
          MATCH (m),(p)
          WHERE strId(m) = $that.data.movie AND strId(p) = $that.data.person
          MERGE (p:Person)-[:ActedDirected]->(m:Movie)
      log-actor-director:
        type: WriteToFile
        path: "ActorDirector.jsonl"
```

My standing query creates a new `ActedDirected` relationship between the Person and Movie nodes, then logs the relationship. 

Four hundred ninety-one actors acted in and directed the same movie in our data set.

```JSON
{
    "data": {
        "Actor": "Clint Eastwood",
        "Movie": "Unforgiven",
        "movieId": "4a6d64c8-9c90-3362-b443-4d2e7b2fb9d1",
        "personId": "4638a820-3b68-3fc7-9fa7-341e876b701e"
    }
}
```

## Conclusion
Phew, we made it through! And we learned a lot along the way. 

* CSV data is streamed into Quine
* Quine can read from external files and streaming providers 
* You can ingest multiple streams at once, movies and reviewers, and combine them into one streaming graph
* Always separate ingest queries using the jobs to be done framework

Quine is open source if you want to run this analysis for yourself. Download a precompiled version or build it yourself from the codebase ([Quine Github] (https://github.com/thatdot/quine)). I published the recipe that I developed at `https://quine.io/recipes`. The page has instructions for downloading the `CSV` files and running the recipe. 

Have a question, suggestion, or improvement? I welcome your feedback! Please drop in to [Quine Slack](https://quine-io.slack.com/) and let me know. I'm always happy to discuss Quine or answer questions.