markhneedham/blog huggingface dataset:
KSQL: Create Stream - extraneous input 'properties'
Date: 2019-05-20 11:43:00
Learn how to create a KSQL stream from a source topic that contains a column with a reserved word.

In my continued playing with the https://www.confluent.io/product/ksql/[KSQL streaming engine for Kafka^], I came across another interesting error while trying to put a stream on top of a topic generated by the https://neo4j-contrib.github.io/neo4j-streams/[Neo4j Streams Library^].

We'll simplify the events being posted on the topic for this blog post, so this is what the events on the topic look like:

[source, javascript]
----
{
   "id":"ABCDEFGHI",
   "properties": {
      "name":"Mark",
      "location":"London"
   }
}
----

We then create a stream on that topic:

[source, sql]
----
CREATE STREAM users_original(
  id varchar,
  properties STRUCT<
    name varchar,
    location varchar
  >
)
WITH(KAFKA_TOPIC='users', value_format='json');
----

But when we execute that query we'll get the following exception:

[source, text]
----
line 3:3: extraneous input 'properties' expecting {'INTEGER', 'DATE', 'TIME', 'TIMESTAMP', 'INTERVAL', 'YEAR', 'MONTH', 'DAY', 'HOUR', 'MINUTE', 'SECOND', 'ZONE', 'PARTITION', 'STRUCT', 'EXPLAIN', 'ANALYZE', 'TYPE', 'SHOW', 'TABLES', 'COLUMNS', 'COLUMN', 'PARTITIONS', 'FUNCTIONS', 'FUNCTION', 'ARRAY', 'MAP', 'SET', 'RESET', 'SESSION', 'IF', IDENTIFIER, DIGIT_IDENTIFIER, QUOTED_IDENTIFIER, BACKQUOTED_IDENTIFIER}
Statement: CREATE STREAM users_original(
----

I thought I might have a trailing comma or colon - two mistakes I've made before - but on this occassion the problem is that `properties` is a reserved word in KSQL.
Luckily it's an https://docs.confluent.io/current/ksql/docs/developer-guide/syntax-reference.html[easy problem to fix^]:

[quote, KSQL Syntax Reference]
____
If the name of a column in your source topic is one of the reserved words in KSQL you can use back quotes to define the column. The same applies to the field names in a STRUCT type. For instance, if in the above example we had another field called Properties, which is a reserved word in KSQL, you can use the following statement to declare your stream:
____

The fixed syntax is below:

[source, sql]
----
CREATE STREAM users_original(
  id varchar,
  `properties` STRUCT<
    name varchar,
    location varchar
  >
)
WITH(KAFKA_TOPIC='users', value_format='json');
----

And if we run that the stream will be created:

[source, text]
----
Message
----------------
Stream created
----------------
----
KSQL: Create Stream - Failed to prepare statement: name is null
Date: 2019-05-19 19:21:00
Learn how to work around a weird KSQL error message.

I've been playing with https://www.confluent.io/product/ksql/[KSQL^] over the weekend and ran into a basic error message that took me a little while to solve.

I was trying to create a stream  over a topic `dummy1`, which is the simplest possible thing you can do with KSQL.
The events posted to `dummy1` are JSON messages containing only an `id` key.
Below is an example of a message posted to the topic:

[source, javascript]
----
{ "id": "ABCDEFGHI"}
----

I tried to create the stream `dummy1_original` over this topic with the following statement:

[source, sql]
----
ksql> CREATE STREAM dummy1_original (id varchar) WITH (kafka_topic='dummy1', value_format="JSON");
Failed to prepare statement: name is null
----

I didn't find any search results to explain this error message, but realised that I'd used double quotes on the `value_format`, which was different to all the examples I'd seen.
I've clearly been doing too much Cypher where you can use single or double quotes to represent string values.

Anyway...let's update our  create statement to use only single quotes:

[source, sql]
----
ksql> CREATE STREAM dummy1_original (id varchar) WITH (kafka_topic='dummy1', value_format='JSON');

 Message
----------------
 Stream created
----------------
----

Success!
We can now query the stream like this:

[source, sql]
----
ksql> SELECT * FROM dummy1_original;
----
Neo4j: keep/filter keys in a map using APOC
Date: 2019-05-12 17:58:00
Learn how to create a node based on a subset of keys in a map.

In this post we'll learn how to write a Cypher query to create a node in Neo4j containing some of the keys from a map.
This post assumes that the https://neo4j.com/developer/neo4j-apoc/[APOC library^] is installed.

We'll start by creating a map that contains data from my twitter profile:

[source, cypher]
----
:param document => {
  id: 14707949, name: "Mark Needham",
  username: "markhneedham", bio: "Developer Relations @neo4j",
  location: "London, United Kingdom", url: "http://www.markhneedham.com",
  join_date: "8 May 2008", join_time: "5:58 PM",
  tweets: 24710, following: 2479, followers: 5054, likes: 1014
};
----

We want to create a `User` node based on this data, but we don't want to use all of the keys in the map.
If we want to remove some keys we can use the `apoc.map.clean` function.
The following function call removes the `join_date` and `join_time` keys:

[source, cypher]
----
WITH ["join_date", "join_time"] AS keysToRemove
RETURN apoc.map.clean($document, keysToRemove, []) AS result
----

image::{{<siteurl>}}/uploads/2019/05/keysToRemove.png[]

But what if we want to keep keys rather than remove them?
I ran `apoc.help('apoc.map')` to see if there were any functions to do this, and as far as I can tell there aren't:

image::{{<siteurl>}}/uploads/2019/05/apochelp.png[]

We can, however, combine two APOC functions to get the desired result.
First up, `apoc.map.values` extracts the values for a list of keys:

[source, cypher]
----
WITH ["name", "username", "bio", "following", "followers"] AS keysToKeep
RETURN apoc.map.values($document, keysToKeep)
----

image::{{<siteurl>}}/uploads/2019/05/values.png[]

And we can then use `apoc.map.fromLists` to reconstruct our map:

[source,cypher]
----
WITH ["name", "username", "bio", "following", "followers"] AS keysToKeep
RETURN apoc.map.fromLists(keysToKeep, apoc.map.values($document, keysToKeep)) AS result
----

image::{{<siteurl>}}/uploads/2019/05/keysToKeep.png[]

This code forms part of the following query that creates a `User` node keyed on the `id` property, and then sets the other properties afterwards:

[source, cypher]
----
WITH ["name", "username", "bio", "following", "followers"] AS keysToKeep
MERGE (u:User {id: $document.id })
SET u += apoc.map.fromLists($keysToKeep, apoc.map.values($document, $keysToKeep))
----

All done!
Neo4j: Conditional WHERE clause with APOC
Date: 2019-07-31 11:08:00
Learn how to write conditional queries using the APOC library.

Sometimes we want to be able to vary our Cypher queries based on the value of a parameter.
I came across such a situation today, and thought I'd share how I solved it using the APOC library.

Let's first setup some sample data:

[source, cypher]
----
UNWIND range(0, 5) AS id
CREATE (:Person {name: "person-" + id})
----

Now, if we want to get all pairs of people, we could write the following query:

[source,cypher]
----
MATCH (p1:Person), (p2:Person)
WHERE id(p1) < id(p2)
RETURN p1.name, p2.name
----

If we run that query we'll see the following output:

.Results
[opts="header",cols="1,1"]
|===
| p1.name | p2.name
|person-0|person-1
|person-0|person-2
|person-0|person-3
|person-0|person-4
|person-0|person-5
|person-1|person-2
|person-1|person-3
|person-1|person-4
|person-1|person-5
|person-2|person-3
|person-2|person-4
|person-2|person-5
|person-3|person-4
|person-3|person-5
|person-4|person-5
|===


But we also want to search for a Person, based on a query parameter, and return the pairs from that Person to everybody else.
Let's see how to do that.

We'll first create a parameter to find a subset of people:

[source, cypher]
----
:param personSubstring => "person-3"
----

And then the following query returns pairs from those people to everybody else:

[source,cypher]
----
MATCH (p1:Person), (p2:Person)
WHERE p1.name CONTAINS $personSubstring AND p2 <> p1
RETURN p1.name, p2.name
----

If we run that query we'll see the following output:

.Results
[opts="header",cols="1,1"]
|===
| p1.name | p2.name
|person-3|person-0
|person-3|person-1
|person-3|person-2
|person-3|person-4
|person-3|person-5
|===

What if we want to combine those queries?
If `personSubstring` contains a value we'll use that, and if not we'll return all pairs of people.

We can use the https://neo4j.com/developer/neo4j-apoc/[APOC Library^]'s `apoc.when` procedure to solve this problem.
This procedure has the following parameters:

* `condition` - a statement that must return a boolean value
* `ifQuery` - the query to execute if the condition is true
* `elseQuery` - the query to execute if the condition is false
* `params` - parameters to pass to the queries

And we can call it like this:

[source,cypher]
----
CALL apoc.when(
  $personSubstring = '',
  'MATCH (p1:Person), (p2:Person) WHERE id(p1) < id(p2) RETURN p1, p2',
  'MATCH (p1:Person), (p2:Person) WHERE p1.name CONTAINS $personSubstring AND p2 <> p1 RETURN p1, p2',
  {personSubstring: $personSubstring})
YIELD value
RETURN value.p1.name, value.p2.name
----

And now we can handle both use cases in one Cypher query.
Neo4j: apoc.load.csv - Neo.ClientError.Statement.SyntaxError: Type mismatch: expected Float, Integer, Number or String but was Any 
Date: 2019-09-05 00:47:00
Learn how to coerce the type of values loaded using APOC's apoc.load.csv procedure.

The https://neo4j.com/docs/labs/apoc/3.5/[Neo4j APOC library^]'s https://neo4j.com/docs/labs/apoc/current/import/load-csv/[Load CSV procedure^] is very useful if you want more control over the import process than the `LOAD CSV` clause allows.
I found myself using it last week to import a CSV file of embeddings, because I wanted to know the line number of the row in the CSV file while importing the data.

I had a file that looked like this, which I put into the `import` directory:

[source,text]
----
$ cat import/data.csv
0.034 0.765 0.452
0.312 0.413 0.789
----

And before I imported it, I added the following entries to my Neo4j Settings file so that I could read locally:

[source,text]
----
apoc.import.file.enabled=true
apoc.import.file.use_neo4j_config=true
----

The following query processes the file, with optional config that indicates that the file doesn't have a header and uses a space as a separator:


[source,cypher]
----
CALL apoc.load.csv("file:///data.csv", {header: false, sep: " "})
YIELD lineNo, map, list
RETURN lineNo, list
----

If we run that query we'll get this result:

.Results
[opts="header",cols="1,2"]
|===
| lineNo | list
|0       | ["0.034","0.765","0.452"]
|1       | ["0.312","0.413","0.789"]
|===

But we want to have each item of the list be a float value rather than a string.
I tried to coerce each of the values using the `toFloat` function:

[source,cypher]
----
CALL apoc.load.csv("file:///data.csv", {header: false, sep: " "})
YIELD lineNo, map, list
RETURN lineNo, [item in list | toFloat(item)] AS list
----

Unfortunately this doesn't quite work, as the following error indicates:

[source,text]
----
Neo.ClientError.Statement.SyntaxError: Type mismatch: expected Float, Integer, Number or String but was Any (line 3, column 40 (offset: 129))
"RETURN lineNo, [item in list | toFloat(item)] AS list"
                                        ^
----

Instead we need to use APOC's `apoc.convert.toFloat` function to do the type coercion.
The following query does the trick:

[source,cypher]
----
CALL apoc.load.csv("file:///data.csv", {header: false, sep: " "})
YIELD lineNo, map, list
RETURN lineNo, [item in list | apoc.convert.toFloat(item)] AS list
----

.Results
[opts="header",cols="1,2"]
|===
| lineNo | list
|0       | [0.034,0.765,0.452]
|1       | [0.312,0.413,0.789]
|===
Python: Add query parameters to a URL
Date: 2019-01-11 09:42:00
Learn how to add query parameters to a URL in Python.

I was recently trying to automate adding a query parameter to a bunch of URLS and came across a neat approach a long way down this https://stackoverflow.com/questions/2506379/add-params-to-given-url-in-python[StackOverflow answer^], that uses the  http://docs.python-requests.org/en/master/api/#requests.PreparedRequest[PreparedRequest^] class from the http://docs.python-requests.org/en/master/[requests library^].

Let's first get the class imported:

[source,cypher]
----
from requests.models import PreparedRequest
req = PreparedRequest()
----

And now let's use use this class to add a query parameter to a URL.
We can do this with the following code:

[source, cypher]
----
url = "http://www.neo4j.com"
params = {'ref':"mark-blog"}
req.prepare_url(url, params)
----

And then we need to access the `url` attribute to see our new URL:

[source, cypher]
----
>>> req.url
'http://www.neo4j.com/?ref=mark-blog'
----

Neat!
We can also use this approach to add parameters to URLs that already have existing ones.
For example, we could update this YouTube URL:

[source, cypher]
----
url = "https://www.youtube.com/watch?v=Y-Wqna-hC2Y&list=RDhlznpxNGFGQ&index=2"
params = {'ref':"mark-blog"}
req.prepare_url(url, params)
----

And let's see what the updated URL looks like:

[source, cypher]
----
>>> req.url
'https://www.youtube.com/watch?v=Y-Wqna-hC2Y&list=RDhlznpxNGFGQ&index=2&ref=mark-blog'
----

I'm sure I'll be using this code snippet in future!
Neo4j: APOC - Caused by: java.io.RuntimeException: Can't read url or key file (No such file or directory)
Date: 2019-01-12 19:05:00
Learn how to import files from the import folder when using the APOC library.

I've been using Neo4j's https://neo4j-contrib.github.io/neo4j-apoc-procedures/[APOC^] library to load some local JSON files this week, and ran into an interesting problem.

The `LOAD CSV` tool assumes that any files you load locally are in the `import` directory, so I've got into the habit of putting my data there.
Let's check what I'm trying to import by opening the import directory:

image::{{<siteurl>}}/uploads/2019/01/import-directory.png[]

What's in there?

image::{{<siteurl>}}/uploads/2019/01/import-directory-contents.png[]

Just the one JSON file needs processing.
If we want to import local files we need to add the following property to our Neo4j configuration file:

[source, text]
----
apoc.import.file.enabled=true
----

If you're using the Neo4j Desktop, you can add this property via the 'Settings' tab:

image::{{<siteurl>}}/uploads/2019/01/Selection_142.png[]

Once we've done that we'll need to restart the database so our new settings will be picked up:

image::{{<siteurl>}}/uploads/2019/01/restart-after-config.png[]

Now let's try to process our JSON file:

[source, cypher]
----
neo4j> CALL apoc.load.json("file:///dummy.json");
Failed to invoke procedure `apoc.load.json`: Caused by: java.lang.RuntimeException: Can't read url or key file:/dummy.json as json: /dummy.json (No such file or directory)
----

Hmm, that didn't work as we expected - it seems to be trying to read the file from the root of the machine rather than from the `import` directory.
It turns out I hadn't RTFM:

image::{{<siteurl>}}/uploads/2019/01/local-files.png[]

Let's update our Neo4j configuration file to add the following property:

[source, text]
----
apoc.import.file.use_neo4j_config=true
----

If we re-run our query we'll see that it now finds and processes the file:

image::{{<siteurl>}}/uploads/2019/01/load-json.png[]
Neo4j: Cypher - Remove consecutive duplicates from a list
Date: 2019-01-12 04:32:00
Learn how to remove consecutive duplicate elements/duplicate neighbours from a list.

I was playing with a dataset this week and wanted to share how I removes duplicate consecutive elements from a list using the Cypher query language.

For simplicity's sake, imagine that we have this list:

[source, cypher]
----
neo4j> return [1,2,3,3,4,4,4,5,3] AS values;
+-----------------------------+
| values                      |
+-----------------------------+
| [1, 2, 3, 3, 4, 4, 4, 5, 3] |
+-----------------------------+
----

We want to remove the duplicate 3's and 4's, such that our end result should be:

[source, cypher]
----
[1,2,3,4,5,3]
----

https://github.com/neo4j-contrib/neo4j-apoc-procedures[APOC]'s `apoc.coll.toSet` doesn't quite do the trick because it removes duplicates regardless of where they appear in the collection:

[source, cypher]
----
neo4j> return apoc.coll.toSet([1,2,3,3,4,4,4,5,3]) AS values;
+-----------------+
| values          |
+-----------------+
| [1, 2, 3, 4, 5] |
+-----------------+
----

Luckily it's quite easy to translate https://stackoverflow.com/a/46977206[Ulf Aslak's Python one liner] to do what we want.
This is the Python version:

[source,python]
----
values = [1,2,3,3,4,4,4,5,3]

>>> [v for i, v in enumerate(values) if i == 0 or v != values[i-1]]
[1, 2, 3, 4, 5, 3]
----

We'll use the https://neo4j.com/docs/cypher-manual/current/functions/list/#functions-range[`range`^] function to iterate over our list and list comprehensions to do the rest.
The following code does the trick:

[source, cypher]
----
neo4j> WITH [1,2,3,3,4,4,4,5,3] AS values
       RETURN [i in range(0, size(values)-1)
               WHERE i=0 OR values[i] <> values[i-1] | values[i] ] AS values;
+--------------------+
| values             |
+--------------------+
| [1, 2, 3, 4, 5, 3] |
+--------------------+
----

You can use this on collections containing nodes, strings, or anything else - I've just used numbers in the example to keep the example simple.
Neo4j: Cypher - Date ranges
Date: 2019-01-13 06:42:00
Learn how to remove consecutive duplicate elements/duplicate neighbours from a list.

As part of a dataset I've been working with this week, I wanted to generate a collection of a range of dates using the Cypher query language.

I've previously used the https://neo4j.com/docs/cypher-manual/current/syntax/temporal/#cypher-temporal-durations[`duration`^] function, which lets you add (or subtract) from a specific date, so I thought I'd start from there.
If we want to find the day after 1st January 2019, we could write the following query:

[source,cypher]
----
neo4j> WITH date("2019-01-01") AS startDate
       RETURN startDate + duration({days: 1}) AS date;
+------------+
| date       |
+------------+
| 2019-01-02 |
+------------+
----

We can extend this code sample to find the next 5 dates from 1st January 2019 by using the `range` function:

[source,cypher]
----
neo4j> WITH date("2019-01-01") AS startDate
       RETURN [day in range(0, 5) | startDate + duration({days: day})]
       AS dates;
+--------------------------------------------------------------------------+
| dates                                                                    |
+--------------------------------------------------------------------------+
| [2019-01-01, 2019-01-02, 2019-01-03, 2019-01-04, 2019-01-05, 2019-01-06] |
+--------------------------------------------------------------------------+
----

That works fine if we know how many days we want to find the range for, but what if we only know the start and end dates for which we want to extract a range?
We can use the `duration.inDays` function to compute a duration between two dates:

[source, cypher]
----
neo4j> RETURN duration.inDays(date("2019-01-01"), date("2019-01-06")) AS difference;
+------------+
| difference |
+------------+
| P0M5DT0S   |
+------------+
----

We can then get a count of the number of days from the `days`attribute:

[source,cypher]
----
neo4j> RETURN duration.inDays(date("2019-01-01"), date("2019-01-06")).days AS days;
+------+
| days |
+------+
| 5    |
+------+
----

We can now update our query where we hard coded the number of days to use this computed value instead:

[source,cypher]
----
neo4j> WITH date("2019-01-01") AS startDate, date("2019-01-06") AS endDate
       WITH startDate, duration.inDays(startDate, endDate).days AS days
       RETURN [day in range(0, days) | startDate + duration({days: day})]
       AS dates;
+--------------------------------------------------------------------------+
| dates                                                                    |
+--------------------------------------------------------------------------+
| [2019-01-01, 2019-01-02, 2019-01-03, 2019-01-04, 2019-01-05, 2019-01-06] |
+--------------------------------------------------------------------------+
----
Pandas: Add row to DataFrame
Date: 2021-05-13 00:44:37
In this post we'll add a new row to a Pandas DataFrame.

Usually when I'm working with Pandas DataFrames I want to add new columns of data, but I recently wanted to add a row to an existing DataFrame.
It turns out there are more than one ways to do that, which we'll explore in this blog post. 

Let's start by importing Pandas into our Python script:

[source, bash]
----
import pandas as pd
----

We'll start from a DataFrame that has two rows and the columns `name` and `age`:


[source, python]
----
df = pd.DataFrame(data=[{"name": "Mark", "age": 37}, {"name": "David", "age": 36}])
----

.DataFrame with no explicit index
[format="csv", options="header"]
|===
include::content/2021/05/13/data/no_index.csv[]
|===

One way to add rows to this DataFrame is by creating a new DataFrame and joining it with our initial DataFrame using the `append` function:

[source, python]
----
to_append = pd.DataFrame([
    {"name": "Ryan", "age": 42},
    {"name": "John", "age": 25}
])
append_df = df.append(to_append)
----

.DataFrame with new row
[format="csv", options="header"]
|===
include::content/2021/05/13/data/no_index_append.csv[]
|===

The index for the new rows start again from 0, so we now have multiple rows with the index 0 and 1.

Another way that we can append a new row is using the `loc` function:

[source, python]
----
df.loc[2] = ["Ryan", 42]
df.loc[3] = ["John", 25]
----

If we take this approach we need to explicitly specify the index of the row.
We can put whatever value we want, but let's create indexes that increment the existing values:

.DataFrame with new row
[format="csv", options="header"]
|===
include::content/2021/05/13/data/no_index_individual.csv[]
|===

What if we have an explicit index set on the DataFrame?
We can convert the `name` column into an index by running the following code:

[source, python]
----
df_with_index = df.set_index("name")
----

.DataFrame with `name` index
[format="csv", options="header"]
|===
include::content/2021/05/13/data/index.csv[]
|===

And now if we want to add a new row the index identifier should be a name instead of a numeric value:

[source, python]
----
df_with_index.loc["Ryan"] = [42]
df_with_index.loc["John"] = [25]
----

.DataFrame with `name` index with new row
[format="csv", options="header"]
|===
include::content/2021/05/13/data/index_with_row.csv[]
|===
Python: Select keys from map/dictionary
Date: 2020-04-27 00:21:00
Learn how to filter a map to select specific keys

In this post we're going to learn how to filter a Python map/dictionary to return a subset of keys or values.
I needed to do this recently while logging some maps that had a lot of keys that I wasn't interested in. 

We'll start with the following map:

[source, python]
----
x = {"a": 1, "b": 2, "c": 3, "d": 4}
{'a': 1, 'b': 2, 'c': 3, 'd': 4}
----

We want to filter this map so that we only have the keys `a` and `c`.
If we just want the values, we can create a list comprehension to do this:

[source,python]
----
>>> [x[key] for key in ["a", "b"]]
[1, 2]
----

But what if we want to return the values as well?
This is where https://markhneedham.com/blog/2013/08/13/python-forlist-comprehensions-and-dictionaries/[dictionary comprehensions^] come in handy.
We can tweak our code as follows:

[source,python]
----
>>> {key: x[key] for key in ["a", "b"]}
{'a': 1, 'b': 2}
----

Or we can iterate over all the entries in the map and filter it that way:

[source,python]
----
>>> {key:value for key,value in x.items() if key in ["a", "b"]}
{'a': 1, 'b': 2}
----

This approach is longer but more flexible.
For example, we could find the keys and values for all entries with a value great than 2 with the following code:

[source,python]
----
>>> {key:value for key,value in x.items() if value > 2}
{'c': 3, 'd': 4}
----