...

Neo4j database contains information about Persons and Movies and relation between them.
For getting information what movies related with person 'Meg Ryan' can be used nex next CQL query:

MATCH (person:Person {name: "Meg Ryan"})-[rel]-(movie) RETURN person, rel, movie

Result of this query will be next:

"person"

"rel"

"movie"

{"name":"Meg Ryan","born":1961}

{"roles":["DeDe","Angelica Graynamore","Patricia Graynamore"]}

{"title":"Joe Versus the Volcano",
"tagline":"A story of love, lava andburning desire.",
"released":1990}

{"name":"Meg Ryan","born":1961}

{"roles":["Sally Albright"]}

{"title":"When Harry Met Sally",
"tagline":"Can two friends sleep toget her and still love each other in the morning?",
"released":1998}

{"name":"Meg Ryan","born":1961}

{"roles":["Kathleen Kelly"]}

{"title":"You've Got Mail",
"tagline":"At odds in life... in love on-line.",
"released":1998}

{"name":"Meg Ryan","born":1961}

{"roles":["Carole"]}

{"title":"Top Gun",
"tagline":"I feel the need, the need for speed.",
"released":1986}

{"name":"Meg Ryan","born":1961}

{"roles":["Annie Reed"]}

{"title":"Sleepless in Seattle",
"tagline":"What if someone you never met, someone you never saw, someone you never 
knew was the only someone for you?",
"released":1993}

Source Splitter

The proposal is to add "Splits Number" Source configuration property, which allows specifying the desired number of splits to divide the query into when reading from Neo4j. 
Fewer splits may be created if the query cannot be divided into the desired number of splits.
Also, we can use '0' as the default value for this configuration property and determine the number of splits according to the number of map tasks (controlled by the "mapreduce.job.maps" property):

public List<InputSplit> getSplits(JobContext job) throws IOException {
 
    ...
 
    int targetNumTasks = job.getConfiguration().getInt(MRJobConfig.NUM_MAPS, 1);
     
    ...

'MATCH ... RETURN COUNT(*)' CQL query can be used in order to get a total number of documents, that will be divided between splits using 'SKIP' and 'LIMIT'

Source Properties

...

The query to use to import data from the Neo4j database.
Query example: 'MATCH (n:Label) RETURN n.property_1, n.property_2'.

...

Field Name which will be used for ordering during splits generation. This is required unless numSplits is set to one.

Source Data Types Mapping

{"name": "n", "type": {
	"type": "record", "name": "n", "fields": [
		{"name": "born", "type": "long"}, 
		{"name": "name", "type": "string"}, 
		{"name": "_id", "type": "long"}, 
		{"name": "_labels", "type": {"type": "array", "items": "string"}}
	]
}}

Other case using CQL for getting data from Neo4j:

MATCH (person:Person {name: "Meg Ryan"})-[rel]-(movie) RETURN person.name AS name, rel.roles AS roles, movie.title AS title

Result of this query will be next:

title

Source Splitter

The proposal is to add "Splits Number" Source configuration property, which allows specifying the desired number of splits to divide the query into when reading from Neo4j. 
Fewer splits may be created if the query cannot be divided into the desired number of splits.
Also, we can use '0' as the default value for this configuration property and determine the number of splits according to the number of map tasks (controlled by the "mapreduce.job.maps" property):

public List<InputSplit> getSplits(JobContext job) throws IOException {
 
    ...
 
    int targetNumTasks = job.getConfiguration().getInt(MRJobConfig.NUM_MAPS, 1);
     
    ...

'MATCH ... RETURN COUNT(*)' CQL query can be used in order to get a total number of documents, that will be divided between splits using 'SKIP' and 'LIMIT'
Example:

• Input query
  

  Code Block
  MATCH (person:Person) RETURN person

• Order By

  Code Block
  person.born

In this case each split will be run next query

MATCH (person:Person) RETURN person ORDER BY person.born SKIP x LIMIT y

where 'x' and 'y' determined for each split based on 'Splits Number' and total counts of records.

Source Properties

Source Data Types Mapping

{"name": "n", "type": {
	"type": "record", "name": "n", "fields": [
		{"name": "born", "type": "long"}, 
		{"name": "name", "type": "string"}, 
       		{"name": "_endIdid", "type": "long"}, 
       		{"name": "_idlabels", "type": {"longtype"}:     "array", "items": "string"}}
	]
}}

{"name": "drr", "type": {
    "type": "record", "name": "drr", "fields": [
        {"name": "duration_startId", "type": "stringlong"},
        {"name": "secondsroles", "type": {"type": "array", "items": "longstring"}},
        {"name": "_type", "type":      {"name": "months", "type": "long"},
        {"name": "days", "type": "long"},
        {"name": "nanoseconds", "type": "int"}
    ]
}}

{"name": "p", "type": {
    "type": "record", "name": "p", "fields": [
        {"name": "crs", "type": "string"},
        {"name": "x", "type": "double"},
        {"name": "y", "type": "double"},
        {"name": "srid", "type": "string"}
    ]
}}

Sink Properties

"string"},
        {"name": "_endId", "type": "long"},
        {"name": "_id", "type": "long"}
    ]
}}

{"name": "dr", "type": {
    "type": "record", "name": "dr", "fields": [
        {"name": "duration", "type": "string"},
        {"name": "seconds", "type": "long"},
        {"name": "months", "type": "long"},
        {"name": "days", "type": "long"},
        {"name": "nanoseconds", "type": "int"}
    ]
}}

{"name": "p", "type": {
    "type": "record", "name": "p", "fields": [
        {"name": "crs", "type": "string"},
        {"name": "x", "type": "double"},
        {"name": "y", "type": "double"},
        {"name": "srid", "type": "string"}
    ]
}}

Sink Properties

Output query additionl information

Output query is based on CQL syntax, but using CQL query with CDAP has several problem:

• neo4j-jdbc-driver can process property values only if it primitive types or arrays thereof.
• difficult to relate the output data to CQL query.

To solve these problems, the following solution was proposed:
Using next structure $(...) for identify place where properties will be inserted.
Example of using $(...):
List of output fields: ["name", "age", "profesion", "company", "rating", "position"]

Sink Data Types Mapping

...