Neo4j/twitter Kafka Connect demo

gistfile1.txt

= Testing the Neo4j Connector with Twitter as a source of data

Caveat: I have never used Neo4j before. I'm familiar with the general concept of a property graph though. 

Environment: https://github.com/rmoff/neo4j-streams/blob/master/kafka-connect-neo4j/docker/docker-compose.yml

== Start here

Create a Twitter source

[source,bash]
----
echo '{
    "name": "twitter_source_avro",
    "config": {
      "tasks.max" : "1",
      "key.converter": "io.confluent.connect.avro.AvroConverter",
      "key.converter.schema.registry.url": "http://schema_registry:8081",
      "value.converter": "io.confluent.connect.avro.AvroConverter",
      "value.converter.schema.registry.url": "http://schema_registry:8081",
      "kafka.delete.topic": "twitter_deletes_avro",
      "twitter.oauth.consumerKey": "get yours from dev.twitter.com",
      "twitter.oauth.consumerSecret": "get yours from dev.twitter.com",
      "twitter.oauth.accessToken": "get yours from dev.twitter.com",
      "twitter.oauth.accessTokenSecret": "get yours from dev.twitter.com",
      "kafka.status.topic": "twitter_avro",
      "connector.class": "com.github.jcustenborder.kafka.connect.twitter.TwitterSourceConnector",
      "process.deletes": true, 
      "filter.keywords": "java"
    }
  }
  ' | http POST localhost:8083/connectors
----

== Stream to Neo4j

[source,bash]
----
echo '{
  "name": "Neo4jSinkConnector-twitter_avro",
  "config": {
    "key.converter":"org.apache.kafka.connect.storage.StringConverter",
    "topics": "twitter_avro",
    "connector.class": "streams.kafka.connect.sink.Neo4jSinkConnector",
    "errors.tolerance": "all",
    "errors.log.enable": true,
    "errors.log.include.messages": true,
    "neo4j.server.uri": "bolt://neo4j:7687",
    "neo4j.authentication.basic.username": "neo4j",
    "neo4j.authentication.basic.password": "connect",
    "neo4j.topic.cypher.twitter_avro": "MERGE (u:User{ID: event.User.Id, Name: event.User.Name,ScreenName: event.User.ScreenName, from: \''AVRO\''}) MERGE (t:Tweet{text: event.Text, User:event.User.Id}) MERGE (u)-[:TWEETED]->(t)"
  }
} ' | http POST localhost:8083/connectors
----