Skip to content

PLUGIN-12: Added Neo4j database Source and Sink plugins. #62

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 4 commits into
base: develop
Choose a base branch
from
Open

PLUGIN-12: Added Neo4j database Source and Sink plugins. #62

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
628956e
PLUGIN-12: Added Neo4j Source plugin.
AntonGensitskiy Oct 24, 2019
6f61c6e
PLUGIN-12: Added Neo4j Sink plugin.
AntonGensitskiy Nov 3, 2019
e22b4c5
PLUGIN-12: Added integration tests.
AntonGensitskiy Nov 6, 2019
e516635
PLUGIN-12: Added 'Driver Name' plugin property.
AntonGensitskiy Dec 6, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 52 additions & 0 deletions neo4j-plugin/docs/Neo4jSink-batchsink.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
# Neo4j Batch Sink


Description
-----------
Write data to Neo4j instance using a configurable CQL query.
Count of created Node and Relation depend on Output Query.


Use Case
--------
The sink is used whenever you need to write to a Neo4j instance.


Properties
----------
**Reference Name:** This will be used to uniquely identify this source for lineage, annotating metadata, etc.

**Driver Name:** Name of the JDBC driver to use.

**Neo4j Host:** Neo4j database host.

**Neo4j Port:** Neo4j database port.

**Output Query:** The query to use to export data to the Neo4j database. Query example: 'CREATE (n:<label_field> $(*))'
or 'CREATE (n:<label_field> $(property_1, property_2))'. Addition information can be found on
https://wiki.cask.co/display/CE/Neo4j+database+plugin

**Username:** User to use to connect to the Neo4j database.

**Password:** Password to use to connect to the Neo4j database.


Data Types Mapping
----------

| CDAP Schema Data Types | Neo4j Data Types | Comment |
| ---------------------- | ------------------------------------- | -------------------------------------------- |
| null | null | |
| array | List | |
| boolean | Boolean | |
| long, int | Integer | |
| double | Float | |
| string | String | |
| bytes | ByteArray | |
| date | Date | |
| time-micros | Time, LocalTime | |
| timestamp-micros | DateTime, LocalDateTime | |
| record | Duration, Point | Depending on record fields |



61 changes: 61 additions & 0 deletions neo4j-plugin/docs/Neo4jSource-batchsource.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
# Neo4j Batch Source


Description
-----------
Reads from a Neo4j instance using a configurable CQL query.
Outputs one record for each row returned by the query.


Use Case
--------
The source is used whenever you need to read from a Neo4j instance.


Properties
----------
**Reference Name:** This will be used to uniquely identify this source for lineage, annotating metadata, etc.

**Driver Name:** Name of the JDBC driver to use.

**Neo4j Host:** Neo4j database host.

**Neo4j Port:** Neo4j database port.

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

**Username:** User to use to connect to the Neo4j database.

**Password:** Password to use to connect to the Neo4j database.

**Splits Number:** The number of splits to generate. If set to one, the orderBy is not needed.

**Order By:** Field Name which will be used for ordering during splits generation. This is required unless numSplits
is set to one and 'ORDER BY' keyword not exist in Input Query.


Data Types Mapping
----------

| Neo4j Data Types | CDAP Schema Data Types | Comment |
| ------------------------------- | ---------------------- | -------------------------------------------------- |
| null | null | |
| List | array | |
| Map | record | |
| Boolean | boolean | |
| Integer | long | |
| Float | double | |
| String | string | |
| ByteArray | bytes | |
| Date | date | |
| Time | time-micros | |
| LocalTime | time-micros | |
| DateTime | timestamp-micros | |
| LocalDateTime | timestamp-micros | |
| Node | record | |
| Relationship | record | |
| Duration | record | |
| Point | record | |
| Path | | |

Binary file added neo4j-plugin/icons/Neo4jSink-batchsink.png
View file
Open in desktop
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Binary file added neo4j-plugin/icons/Neo4jSource-batchsource.png
View file
Open in desktop
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
122 changes: 122 additions & 0 deletions neo4j-plugin/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,122 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
Copyright © 2019 CDAP

Licensed under the Apache License, Version 2.0 (the "License"); you may not
use this file except in compliance with the License. You may obtain a copy of
the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
License for the specific language governing permissions and limitations under
the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>database-plugins</artifactId>
<groupId>io.cdap.plugin</groupId>
<version>1.3.0-SNAPSHOT</version>
</parent>

<name>Neo4j plugin</name>
<artifactId>neo4j-plugin</artifactId>
<version>1.3.0-SNAPSHOT</version>
<modelVersion>4.0.0</modelVersion>

<dependencies>
<dependency>
<groupId>io.cdap.cdap</groupId>
<artifactId>cdap-etl-api</artifactId>
</dependency>
<dependency>
<groupId>io.cdap.plugin</groupId>
<artifactId>database-commons</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>io.cdap.plugin</groupId>
<artifactId>hydrator-common</artifactId>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
</dependency>
<dependency>
<groupId>org.neo4j</groupId>
<artifactId>neo4j-jdbc-driver</artifactId>
<version>3.4.0</version>
</dependency>

<dependency>
<groupId>io.cdap.plugin</groupId>
<artifactId>database-commons</artifactId>
<version>${project.version}</version>
<type>test-jar</type>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.cdap.cdap</groupId>
<artifactId>hydrator-test</artifactId>
</dependency>
<dependency>
<groupId>io.cdap.cdap</groupId>
<artifactId>cdap-data-pipeline</artifactId>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
</dependency>
<dependency>
<groupId>io.cdap.cdap</groupId>
<artifactId>cdap-api</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.jetbrains</groupId>
<artifactId>annotations</artifactId>
<version>RELEASE</version>
<scope>compile</scope>
</dependency>
</dependencies>

<build>
<plugins>
<plugin>
<groupId>org.apache.felix</groupId>
<artifactId>maven-bundle-plugin</artifactId>
<version>3.3.0</version>
<extensions>true</extensions>
<configuration>
<instructions>
<_exportcontents>
io.cdap.plugin.neo4j.*;
org.apache.commons.logging.*;
org.codehaus.jackson.*
</_exportcontents>
<Embed-Dependency>*;inline=false;scope=compile</Embed-Dependency>
<Embed-Transitive>true</Embed-Transitive>
<Embed-Directory>lib</Embed-Directory>
</instructions>
</configuration>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>bundle</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>io.cdap</groupId>
<artifactId>cdap-maven-plugin</artifactId>
</plugin>
</plugins>
</build>

</project>
67 changes: 67 additions & 0 deletions neo4j-plugin/src/main/java/io/cdap/plugin/neo4j/Neo4jConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
/*
* Copyright © 2019 Cask Data, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
* use this file except in compliance with the License. You may obtain a copy of
* the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations under
* the License.
*/

package io.cdap.plugin.neo4j;

import io.cdap.cdap.api.data.schema.Schema;

import java.util.Arrays;
import java.util.Collections;
import java.util.List;

/**
* Neo4j constants.
*/
public final class Neo4jConstants {
public static final String NEO4J_CONNECTION_STRING_FORMAT = "jdbc:neo4j:bolt://%s:%s/?username=%s,password=%s";
public static final String NAME_REFERENCE_NAME = "referenceName";
public static final String NAME_DRIVER_NAME = "jdbcPluginName";
public static final String NAME_HOST_STRING = "neo4jHost";
public static final String NAME_PORT_STRING = "neo4jPort";
public static final String NAME_USERNAME = "username";
public static final String NAME_PASSWORD = "password";

public static final String OUTPUT_QUERY = "cdap.neo4j.output.query";

public static final List<Schema.Field> DURATION_RECORD_FIELDS = Collections.unmodifiableList(Arrays.asList(
Schema.Field.of("duration", Schema.of(Schema.Type.STRING)),
Schema.Field.of("seconds", Schema.of(Schema.Type.LONG)),
Schema.Field.of("months", Schema.of(Schema.Type.LONG)),
Schema.Field.of("days", Schema.of(Schema.Type.LONG)),
Schema.Field.of("nanoseconds", Schema.of(Schema.Type.INT))
));

public static final List<Schema.Field> POINT_2D_RECORD_FIELDS = Collections.unmodifiableList(Arrays.asList(
Schema.Field.of("srid", Schema.of(Schema.Type.INT)),
Schema.Field.of("x", Schema.of(Schema.Type.DOUBLE)),
Schema.Field.of("y", Schema.of(Schema.Type.DOUBLE))
));

public static final List<Schema.Field> POINT_3D_RECORD_FIELDS = Collections.unmodifiableList(Arrays.asList(
Schema.Field.of("srid", Schema.of(Schema.Type.INT)),
Schema.Field.of("x", Schema.of(Schema.Type.DOUBLE)),
Schema.Field.of("y", Schema.of(Schema.Type.DOUBLE)),
Schema.Field.of("z", Schema.of(Schema.Type.DOUBLE))
));

public static final List<String> NEO4J_SYS_FIELDS = Collections.unmodifiableList(Arrays.asList(
"_id",
"_labels",
"_type",
"_startId",
"_endId"
));
}
Loading