Streaming triple changes from stardog to kafka

s.ar...@geophy.com

unread,

Oct 4, 2016, 8:03:01 AM10/4/16

to Stardog

Does stardog support any change data capture mechanism that allow me to send changes in the database to another system (e.g kafka)?

The use case where this is important is to log data for auditing or send data to be indexed into an external system (mongodb or elasticsearch).

I would like to be able to collect any change that happens in the database into a queue system. Anyway to do that with stardog?

Zachary Whitley

unread,

Oct 4, 2016, 9:06:32 AM10/4/16

to Stardog

Try taking a look at Stardog versioning http://docs.stardog.com/#_versioning

--
-- --
You received this message because you are subscribed to the C&P "Stardog" group.
To post to this group, send email to sta...@clarkparsia.com
To unsubscribe from this group, send email to
stardog+unsubscribe@clarkparsia.com
For more options, visit this group at
http://groups.google.com/a/clarkparsia.com/group/stardog?hl=en
---
You received this message because you are subscribed to the Google Groups "Stardog" group.
To unsubscribe from this group and stop receiving emails from it, send an email to stardog+unsubscribe@clarkparsia.com.

Samur Araujo

unread,

Oct 4, 2016, 9:21:28 AM10/4/16

to sta...@clarkparsia.com

Hi Zachary, thank you for the pointer. Although a very nice feature to have, I do not see how It would help.

We have many applications writing to the database. I do not see how versioning would help here.

What we have done in mysql is to read the binlog and the send the changes to a queue.The binlog contains any change happened into the database within a window of time. This works perfectly! Notice we do not need versioning but send changes in real time to another system.

We need the same capability on a triplestore.

You received this message because you are subscribed to a topic in the Google Groups "Stardog" group.
To unsubscribe from this topic, visit https://groups.google.com/a/clarkparsia.com/d/topic/stardog/7b7f1R43Rwk/unsubscribe.
To unsubscribe from this group and all its topics, send an email to stardog+unsubscribe@clarkparsia.com.

--

Senior Data Scientist

Geophy

www.geophy.com

Nieuwe Plantage 54-55
2611XK Delft
+31 (0)70 7640725

1 Fore Street
EC2Y 9DT London
+44 (0)20 37690760

Kendall Clark

unread,

Oct 4, 2016, 9:31:22 AM10/4/16

to sta...@clarkparsia.com

Depending on yr requirements, you might be able to create a reader for Stardog's audit logging and pick out events you need there.

http://docs.stardog.com/#_access_audit_logging

Cheers,

Kendall

stardog+u...@clarkparsia.com

Zachary Whitley

unread,

Oct 4, 2016, 9:46:21 AM10/4/16

to Stardog

Take a quick look specifically at http://docs.stardog.com/#_viewing_diffs

"Diffs are represented as SPARQL Update queries so that they may be used as a kind of graph patch."

Evren Sirin

unread,

Oct 4, 2016, 11:25:00 PM10/4/16

to Stardog

Versioning would give you the exact set of changes for each transaction but you'd need to poll for new versions yourself and it might be a little heavy-weight for your use case because all the changes are indexed in the versioning database internally.

An alternative solution would be to use Stardog's extensible transaction framework for attaching a listener and processing transaction data as transaction events are fired. This mechanism is not very user-friendly at this point and requires a little bit boilerplate code but there is a simple example that I now pushed to the stardog-examples repo [1]. In this example there is only one function [2] that you need to modify if you want to send the data to an external system.

For installing this extension, check out the repo, build the jar, create a server/ext directory under the stardog installation directory, and copy the jar file:

$ ./gradlew :examples:listener:jar

$ mkdir <stardog-path>/server/ext/

$ cp examples/listener/build/libs/listener-0.0.3.jar <stardog-path>/server/ext/

After these steps you need to (re)start the Stardog server and you should see some output in stardog.log as changes occur. You can customize the functionality in the example and redeploy the jar as needed.

Best,

Evren

[1] https://github.com/stardog-union/stardog-examples/tree/develop/examples/listener

[2] https://github.com/stardog-union/stardog-examples/blob/develop/examples/listener/main/src/com/complexible/stardog/examples/listener/ListenerConnectableConnection.java#L165

Samur Araujo

unread,

Oct 5, 2016, 3:23:36 AM10/5/16

to sta...@clarkparsia.com

Thank you Evren!

This direction seems promising. I will explore it further.

Best,

Samur

On 5 October 2016 at 05:24, Evren Sirin <ev...@stardog.com> wrote:

Versioning would give you the exact set of changes for each transaction but you'd need to poll for new versions yourself and it might be a little heavy-weight for your use case because all the changes are indexed in the versioning database internally.

An alternative solution would be to use Stardog's extensible transaction framework for attaching a listener and processing transaction data as transaction events are fired. This mechanism is not very user-friendly at this point and requires a little bit boilerplate code but there is a simple example that I now pushed to the stardog-examples repo [1]. In this example there is only one function [2] that you need to modify if you want to send the data to an external system.

For installing this extension, check out the repo, build the jar, create a server/ext directory under the stardog installation directory, and copy the jar file:

$ ./gradlew :examples:listener:jar
$ mkdir <stardog-path>/server/ext/
$ cp examples/listener/build/libs/listener-0.0.3.jar <stardog-path>/server/ext/

After these steps you need to (re)start the Stardog server and you should see some output in stardog.log as changes occur. You can customize the functionality in the example and redeploy the jar as needed.

Best,
Evren

[1] https://github.com/stardog-union/stardog-examples/tree/develop/examples/listener
[2] https://github.com/stardog-union/stardog-examples/blob/develop/examples/listener/main/src/com/complexible/stardog/examples/listener/ListenerConnectableConnection.java#L165

s.ar...@geophy.com

unread,

Jan 9, 2017, 11:06:51 AM1/9/17

to Stardog

Hi Envre, I followed the steps below but it does not worked.

I did all you mentioned + restart stardog.

Does it works for stardog 4.2.2?

Is there a way to see if the listener was loaded?

stardog+u...@clarkparsia.com

For more options, visit this group at
http://groups.google.com/a/clarkparsia.com/group/stardog?hl=en
---
You received this message because you are subscribed to the Google Groups "Stardog" group.

To unsubscribe from this group and stop receiving emails from it, send an email to stardog+u...@clarkparsia.com.

--
-- --
You received this message because you are subscribed to the C&P "Stardog" group.
To post to this group, send email to sta...@clarkparsia.com
To unsubscribe from this group, send email to

stardog+u...@clarkparsia.com

For more options, visit this group at
http://groups.google.com/a/clarkparsia.com/group/stardog?hl=en
---
You received this message because you are subscribed to a topic in the Google Groups "Stardog" group.
To unsubscribe from this topic, visit https://groups.google.com/a/clarkparsia.com/d/topic/stardog/7b7f1R43Rwk/unsubscribe.

To unsubscribe from this group and all its topics, send an email to stardog+u...@clarkparsia.com.

--
Senior Data Scientist
Geophy
www.geophy.com

Nieuwe Plantage 54-55
2611XK Delft
+31 (0)70 7640725
1 Fore Street
EC2Y 9DT London
+44 (0)20 37690760

--
-- --
You received this message because you are subscribed to the C&P "Stardog" group.
To post to this group, send email to sta...@clarkparsia.com
To unsubscribe from this group, send email to

stardog+u...@clarkparsia.com

For more options, visit this group at
http://groups.google.com/a/clarkparsia.com/group/stardog?hl=en

--
-- --
You received this message because you are subscribed to the C&P "Stardog" group.
To post to this group, send email to sta...@clarkparsia.com
To unsubscribe from this group, send email to

stardog+u...@clarkparsia.com

For more options, visit this group at
http://groups.google.com/a/clarkparsia.com/group/stardog?hl=en
---
You received this message because you are subscribed to the Google Groups "Stardog" group.

To unsubscribe from this group and stop receiving emails from it, send an email to stardog+u...@clarkparsia.com.

--
-- --
You received this message because you are subscribed to the C&P "Stardog" group.
To post to this group, send email to sta...@clarkparsia.com
To unsubscribe from this group, send email to

stardog+u...@clarkparsia.com

Zachary Whitley

unread,

Jan 9, 2017, 2:12:37 PM1/9/17

to Stardog

On Mon, Jan 9, 2017 at 11:06 AM, <s.ar...@geophy.com> wrote:

Hi Envre, I followed the steps below but it does not worked.

I did all you mentioned + restart stardog.

There were no problems with the build? No errors? No warnings?

Does it works for stardog 4.2.2?

Is there a way to see if the listener was loaded?

The first thing you should see is "Transaction started for database..." in the logs. Are you not seeing anything? No errors? No warnings? No transaction change info?

Stupid question but I'm assuming you're writing some new data to the database before checking the logs? Did you put the jar in the correct location? $STARDOG/server/ext (need to create ext dir). Does the user you're running Stardog as have access to the server/ext directory and the jar?

stardog+unsubscribe@clarkparsia.com

For more options, visit this group at
http://groups.google.com/a/clarkparsia.com/group/stardog?hl=en
---
You received this message because you are subscribed to the Google Groups "Stardog" group.

To unsubscribe from this group and stop receiving emails from it, send an email to stardog+unsubscribe@clarkparsia.com.

Samur Araujo

unread,

Jan 10, 2017, 3:49:58 AM1/10/17

to sta...@clarkparsia.com

Hi, I started over and this time it produces the following exception when I try to add/remove data.

I did not change the code in the repo. I've just compiled it and copied the jar to server/ext.

ERROR 2017-01-10 09:43:28,615 [Stardog.Executor-59] com.complexible.common.protocols.server.rpc.ServerHandler:exceptionCaught(428): exceptionCaughtServerHandler

java.lang.AbstractMethodError: Method com/complexible/stardog/examples/listener/ListenerConnectableConnection.getQueryRewritings()Ljava/util/Set; is abstract

at com.complexible.stardog.examples.listener.ListenerConnectableConnection.getQueryRewritings(ListenerConnectableConnection.java) ~[listener-0.0.3.jar:?]

at com.complexible.stardog.db.ConnectableConnectionCollection.lambda$getQueryRewritings$39(ConnectableConnectionCollection.java:155) ~[stardog-4.2.2.jar:?]

at java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:267) ~[?:1.8.0_91]

at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374) ~[?:1.8.0_91]

at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481) ~[?:1.8.0_91]

at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471) ~[?:1.8.0_91]

at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708) ~[?:1.8.0_91]

at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:1.8.0_91]

at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499) ~[?:1.8.0_91]

at com.complexible.stardog.db.ConnectableConnectionCollection.getQueryRewritings(ConnectableConnectionCollection.java:155) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.db.DatabaseConnectionImpl.getQueryRewritings(DatabaseConnectionImpl.java:198) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.db.DelegatingConnectableConnection.getQueryRewritings(DelegatingConnectableConnection.java:84) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.plan.cache.SimplePlanCache$PlanCacheContext.<init>(SimplePlanCache.java:164) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.plan.cache.SimplePlanCache.get(SimplePlanCache.java:54) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.plan.eval.ExecutablePlanFactory.createOptimized(ExecutablePlanFactory.java:101) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.plan.eval.QueryEngine.getExecutablePlan(QueryEngine.java:240) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.plan.eval.QueryEngine.executeUpdate(QueryEngine.java:162) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.query.DefaultQueryFactory$UpdateQueryImpl.execute(DefaultQueryFactory.java:263) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.query.DefaultQueryFactory$UpdateQueryImpl.execute(DefaultQueryFactory.java:242) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.query.DelegatingUpdateQuery.execute(DelegatingUpdateQuery.java:139) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.db.DatabaseImpl$TimerPauseUpdate.execute(DatabaseImpl.java:1571) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.db.DatabaseImpl$TimerPauseUpdate.execute(DatabaseImpl.java:1559) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.query.DelegatingUpdateQuery.execute(DelegatingUpdateQuery.java:139) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.StardogKernel$SecuredUpdateQuery.execute(StardogKernel.java:3772) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.StardogKernel$SecuredUpdateQuery.execute(StardogKernel.java:3759) ~[stardog-4.2.2.jar:?]

at com.complexible.stardog.protocols.server.SPECServerFunction.query(SPECServerFunction.java:526) ~[stardog-protocols-spec-server-4.2.2.jar:?]

at com.complexible.stardog.protocols.server.SPECServerFunction.handleMessage(SPECServerFunction.java:152) ~[stardog-protocols-spec-server-4.2.2.jar:?]

at com.complexible.stardog.protocols.http.annex.QueryPanelEndpoint.handleMessage(QueryPanelEndpoint.java:89) ~[stardog-webconsole-annex-4.2.2.jar:?]

at com.complexible.common.protocols.server.rpc.ServerHandler.lambda$handleMessage$1(ServerHandler.java:337) ~[stardog-protocols-api-server-4.2.2.jar:?]

at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_91]

at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_91]

at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_91]

at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_91]

at java.lang.Thread.run(Thread.java:745) [?:1.8.0_91]

ERROR 2017-01-10 09:43:28,616 [StardogServer.WorkerGroup-1] com.complexible.stardog.protocols.http.server.HttpMessageEncoder:createErrorResponse(293): The result encoder received an error message it could not encode, error was: