Database Feature DATABASESINK

Post Reply
Steffenadensis
Posts: 31
Joined: Tue Jul 29, 2014 11:42 am

Database Feature DATABASESINK

Post by Steffenadensis » Wed Jul 30, 2014 3:44 pm

At the Moment, when iam Typing:

Code: Select all

RubyintoDbs := DATABASESINK({Tableschema=['varchar(128)', 'INTEGER', 'varchar(128)', 'INTEGER', 'varchar(128)', 
'FLOAT', 'BIGINT', 'BIGINT'], connection='con1', table='Ruby', drop='true'}, 
ACCESS({source='source',
wrapper='GenericPush',
transport='TcpServer',
protocol='simplecsv',
dataHandler='Tuple',
options=[['port', '4000'],['delimiter','/']],
schema=[
['a','STRING'],
['b','INTEGER'],
['c','STRING'],
['d','INTEGER'],
['e','STRING'],
['f','FLOAT'],
['g','LONG'],
['h','LONG']
]}))
it works how i expact, but when i type the documented Syntax of http://odysseus.offis.uni-oldenburg.de/ ... YSSEUS.pdf :

Code: Select all

RubyintoDbs := DATABASESINK({connection='con1', table='Ruby', drop='true'}, 
ACCESS({source='source',
wrapper='GenericPush',
transport='TcpServer',
protocol='simplecsv',
dataHandler='Tuple',
options=[['port', '4000'],['delimiter','/']],
schema=[
['a','STRING'],
['b','INTEGER'],
['c','STRING'],
['d','INTEGER'],
['e','STRING'],
['f','FLOAT'],
['g','LONG'],
['h','LONG']
]}))
It do not work and an Error pops up:
java.lang.NullPointerException

So the Documentation should be updated, about what is possible.
Last edited by Steffenadensis on Thu Jul 31, 2014 1:49 pm, edited 1 time in total.
Kind regards

T. Steffen

Operating system: Debian GNU/Linux 7.6 (wheezy)

User avatar
Marco Grawunder
Posts: 272
Joined: Tue Jul 29, 2014 10:29 am
Location: Oldenburg, Germany
Contact:

Re: Database Feature DATABASESINK

Post by Marco Grawunder » Wed Jul 30, 2014 4:31 pm

Both. Error is fixed .... to much work in parallel ;)

Steffenadensis
Posts: 31
Joined: Tue Jul 29, 2014 11:42 am

Re: Database Feature DATABASESINK

Post by Steffenadensis » Thu Jul 31, 2014 2:16 pm

Hey,

i just recognized that the Databasesink is not working correct at the moment.
So it should insert 700 Tuple into AdensisDataStreams.Ruby, but there appear 1000, and that is the size of Bachsize.

So i selected the table, and found out that Odysseus inserts the 700 Tuple and then Startet again until the 1000 is reached.
So in the Table there are duplicatet Tuples, like all h < 300

( Select count(h) from Ruby where h < 300; )

So, iam typing into Odysseus:

Code: Select all

#PARSER CQL
#QUERY
CREATE DATABASE CONNECTION con1 AS mysql TO AdensisDataStreams AT X.X.3.67 : 3306 WITH USER
adensis PASSWORD XXX NO_LAZY_CONNECTION_CHECK

#PARSER PQL
#RUNQUERY
RubyintoDbs := DATABASESINK({Tableschema=['varchar(128)', 'Integer', 'varchar(128)', 'Integer', 'varchar(128)',
 'FLOAT', 'LONG', 'INTEGER' ], Batchsize=1000, 
connection='con1', 
table='Ruby', drop='true'},
ACCESS({source='RubyintoDbs',
wrapper='GenericPush',
transport='TcpServer',
protocol='simplecsv',
dataHandler='Tuple',
options=[['port', '4000'], ['delimiter','/']],
schema=[
['a','STRING'],
['b','INTEGER'],
['c','STRING'],
['d','INTEGER'],
['e','STRING'],
['f','FLOAT'],
['g','LONG'],
['h','LONG']
]}))

#PARSER CQL
#RUNQUERY
Select * from RubyintoDbs;
Source (Ruby code):

Code: Select all

connection = TCPSocket.new(hostname, port)
i = 0
700.times {
	begin
		#connection.print "Str/2323/dsd/234/Hallo/2.4/12345678986/" + i.to_s + "*"
		connection.puts "Str/2323/dsd/234/Hallo/2.4/12345678986/" + i.to_s
		i+=1
	rescue Exception=>e
		puts "counlt not send"
		connection.close
	end
}
Kind regards

T. Steffen

Operating system: Debian GNU/Linux 7.6 (wheezy)

Steffenadensis
Posts: 31
Joined: Tue Jul 29, 2014 11:42 am

Re: Database Feature DATABASESINK

Post by Steffenadensis » Thu Jul 31, 2014 2:26 pm

Batchsize=400 also do not work for 700 Tuples.
Just 400 in the Database.

When i set batchsize = 350, it works.
Kind regards

T. Steffen

Operating system: Debian GNU/Linux 7.6 (wheezy)

Steffenadensis
Posts: 31
Joined: Tue Jul 29, 2014 11:42 am

Re: Database Feature DATABASESINK

Post by Steffenadensis » Thu Jul 31, 2014 2:36 pm

so, 11 times send by batchsize = 10 do work neither, looks like the batchsize must be divisible.

So a Timeout, which allows to send the last not full batchsizebuffer would be great.
Kind regards

T. Steffen

Operating system: Debian GNU/Linux 7.6 (wheezy)

User avatar
Marco Grawunder
Posts: 272
Joined: Tue Jul 29, 2014 10:29 am
Location: Oldenburg, Germany
Contact:

Re: Database Feature DATABASESINK

Post by Marco Grawunder » Thu Jul 31, 2014 2:50 pm

Could have been a race condition? Should be fixed now.

Remark: The operator always wait for batchsize tuples or for the close of the query. I think that here both conditions came together...

In my tests it works now as expected.

User avatar
Marco Grawunder
Posts: 272
Joined: Tue Jul 29, 2014 10:29 am
Location: Oldenburg, Germany
Contact:

Re: Database Feature DATABASESINK

Post by Marco Grawunder » Thu Jul 31, 2014 3:29 pm

:-D
New paramter: BATCHTIMEOUT (Time in ms)

Post Reply

Who is online

Users browsing this forum: No registered users and 1 guest