### Author:  [@wwalker](https://discuss.elastic.co/u/wwalker)
#### Post date: [December 6, 2021, 6:56am UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/1 "2021-12-06T06:56:48Z")
I'm querying a database to get questions asked in a ticket. The query asks for all questions asked by the ticket id. So a single query may return multiple rows. I'd like to be able to parse them all into separate fields for the record. So the query and results would be something like:
`Select question,answer,ticketID from tickets`
| ticketID | question | answer |
| --- | --- | --- |
| 2114 | 2 | tonight |
| 2114 | 4 | tomorrow |
| 2114 | 0 | never |
Then they'd look something like below in Elasticsearch
| field | value |
| --- | --- |
| question.0 | never |
| question.2 | tonight |
| question.4 | tomorrow |
I would assume the JDBC Stream filter would take the results and put them in an array. Would I just use mutate to split them out into individual fields?
---
### Author:  [@wwalker](https://discuss.elastic.co/u/wwalker)
#### Post date: [December 6, 2021, 4:34pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/2 "2021-12-06T16:34:59Z")
Went ahead and ran the query and some good and bad things happened. First, the results were stuck in an object, I believe this to be good and can use the json filter to parse it.
```auto
{
"integeranswer": null,
"questionid": 2,
"dateanswer": null,
"textanswer": null,
"decimalanswer": null,
"booleananswer": true
},
{
"integeranswer": null,
"questionid": 4,
"dateanswer": null,
"textanswer": null,
"decimalanswer": null,
"booleananswer": false
}
```
However, when I configured the json filter, only specifying the target, I get an error and the pipeline shuts down. Any ideas what this error means?
```auto
[logstash.javapipeline] Pipeline worker error, the pipeline will be stopped {:pipeline_id=>"example", :error=>"Missing Converter handling for full class name=org.jruby.ext.date.RubyDateTime, simple name=RubyDateTime", :exception=>Java::OrgLogstash::MissingConverterException, :backtrace=>["org.logstash.Valuefier.fallbackConvert(Valuefier.java:118)", "org.logstash.Valuefier.convert(Valuefier.java:96)", "org.logstash.ConvertedMap$1.visit(ConvertedMap.java:55)", "org.logstash.ConvertedMap$1.visit(ConvertedMap.java:49)", "org.jruby.RubyHash.visitLimited(RubyHash.java:698)", "org.jruby.RubyHash.visitAll(RubyHash.java:683)", "org.logstash.ConvertedMap.newFromRubyHash(ConvertedMap.java:89)", "org.logstash.ConvertedMap.newFromRubyHash(ConvertedMap.java:84)", "org.logstash.Valuefier.lambda$initConverters$12(Valuefier.java:171)", "org.logstash.Valuefier.convert(Valuefier.java:94)", "org.logstash.ConvertedList.newFromRubyArray(ConvertedList.java:64)", "org.logstash.Valuefier.lambda$initConverters$16(Valuefier.java:183)", "org.logstash.Valuefier.convert(Valuefier.java:94)", "org.logstash.ext.JrubyEventExtLibrary$RubyEvent.safeValueifierConvert(JrubyEventExtLibrary.java:355)", "org.logstash.ext.JrubyEventExtLibrary$RubyEvent.ruby_set_field(JrubyEventExtLibrary.java:121)", "D_3a_.Logstash.$7_dot_13_dot_3.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_integration_minus_jdbc_minus_5_dot_0_dot_7.lib.logstash.filters.jdbc_streaming.RUBY$method$process_event$0(D:/Logstash/7.13.3/vendor/bundle/jruby/2.5.0/gems/logstash-integration-jdbc-5.0.7/lib/logstash/filters/jdbc_streaming.rb:163)", "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:119)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:175)", "org.jruby.ir.targets.InvokeSite.fail(InvokeSite.java:290)", "D_3a_.Logstash.$7_dot_13_dot_3.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_integration_minus_jdbc_minus_5_dot_0_dot_7.lib.logstash.filters.jdbc_streaming.RUBY$method$filter$0(D:/Logstash/7.13.3/vendor/bundle/jruby/2.5.0/gems/logstash-integration-jdbc-5.0.7/lib/logstash/filters/jdbc_streaming.rb:136)", "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:106)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:140)", "org.jruby.ir.targets.InvokeSite.fail(InvokeSite.java:267)", "D_3a_.Logstash.$7_dot_13_dot_3.logstash_minus_core.lib.logstash.filters.base.RUBY$method$do_filter$0(D:/Logstash/7.13.3/logstash-core/lib/logstash/filters/base.rb:159)", "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:106)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:140)", "org.jruby.ir.targets.InvokeSite.fail(InvokeSite.java:267)", "D_3a_.Logstash.$7_dot_13_dot_3.logstash_minus_core.lib.logstash.filters.base.RUBY$block$multi_filter$1(D:/Logstash/7.13.3/logstash-core/lib/logstash/filters/base.rb:178)", "org.jruby.runtime.CompiledIRBlockBody.yieldDirect(CompiledIRBlockBody.java:148)", "org.jruby.runtime.BlockBody.yield(BlockBody.java:106)", "org.jruby.runtime.Block.yield(Block.java:184)", "org.jruby.RubyArray.each(RubyArray.java:1809)", "D_3a_.Logstash.$7_dot_13_dot_3.logstash_minus_core.lib.logstash.filters.base.RUBY$method$multi_filter$0(D:/Logstash/7.13.3/logstash-core/lib/logstash/filters/base.rb:175)", "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:106)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:140)", "org.jruby.internal.runtime.methods.DynamicMethod.call(DynamicMethod.java:200)", "org.logstash.config.ir.compiler.FilterDelegatorExt.doMultiFilter(FilterDelegatorExt.java:127)", "org.logstash.config.ir.compiler.AbstractFilterDelegatorExt.multiFilter(AbstractFilterDelegatorExt.java:134)", "org.logstash.generated.CompiledDataset9.compute(Unknown Source)", "org.logstash.generated.CompiledDataset4.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset6.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset6.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset5.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset4.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset6.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset6.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset6.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset6.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset6.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset6.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset6.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset6.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset6.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset6.compute(Unknown Source)", "org.logstash.generated.CompiledDataset2.compute(Unknown Source)", "org.logstash.generated.CompiledDataset5.compute(Unknown Source)", "org.logstash.config.ir.CompiledPipeline$CompiledUnorderedExecution.compute(CompiledPipeline.java:329)", "org.logstash.config.ir.CompiledPipeline$CompiledUnorderedExecution.compute(CompiledPipeline.java:323)", "org.logstash.execution.WorkerLoop.run(WorkerLoop.java:87)", "jdk.internal.reflect.GeneratedMethodAccessor131.invoke(Unknown Source)", "java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)", "java.base/java.lang.reflect.Method.invoke(Method.java:566)", "org.jruby.javasupport.JavaMethod.invokeDirectWithExceptionHandling(JavaMethod.java:441)", "org.jruby.javasupport.JavaMethod.invokeDirect(JavaMethod.java:305)", "org.jruby.java.invokers.InstanceMethodInvoker.call(InstanceMethodInvoker.java:32)", "D_3a_.Logstash.$7_dot_13_dot_3.logstash_minus_core.lib.logstash.java_pipeline.RUBY$block$start_workers$5(D:/Logstash/7.13.3/logstash-core/lib/logstash/java_pipeline.rb:295)", "org.jruby.runtime.CompiledIRBlockBody.callDirect(CompiledIRBlockBody.java:138)", "org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:58)", "org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:52)", "org.jruby.runtime.Block.call(Block.java:139)", "org.jruby.RubyProc.call(RubyProc.java:318)", "org.jruby.internal.runtime.RubyRunnable.run(RubyRunnable.java:105)", "java.base/java.lang.Thread.run(Thread.java:834)"], :thread=>"#"}
```
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [December 6, 2021, 6:12pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/3 "2021-12-06T18:12:05Z")
A jdbc\_streaming filter returns an array of hashes (it has to be an array because the result set may have multiple rows, and they have to be hashes because you can fetch multiple columns). You do not need to parse this as JSON.
You can use a split filter to separate the array into multiple events.
---
### Author:  [@wwalker](https://discuss.elastic.co/u/wwalker)
#### Post date: [December 6, 2021, 7:53pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/4 "2021-12-06T19:53:23Z")
I configured it to use Split...but it looks like that's going to produce unfavorable results. The test record I've done it on only have one question, but it appears as though it will end up creating an array. Is it possible to set the target using split to something like `target => "[question][%{questionid}]"`? The intent would be to stick all the fields in a separate object under each question id...so like:
questionid2.integeranswer
questionid2.dateanswer
questionid2.textanswer
questionid2.decimalanswer
questionid2.booleananswer
questionid4.integeranswer
questionid4.dateanswer
questionid4.textanswer
questionid4.decimalanswer
questionid4.booleananswer
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [December 6, 2021, 8:40pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/5 "2021-12-06T20:40:25Z")
> [@wwalker](#):
>
> Is it possible to set the target using split to something like `target => "[question][%{questionid}]"` ?
No, but you can reformat the array of hashes and change
```
[ { "questionid": 2, "integeranswer": null, ... },
{ "questionid": 4, "integeranswer": null, ... } ]
```
into
```
[ { "questionid2": { "integeranswer": null, ... } },
{ "questionid4": {"integeranswer": null, ... } } ]
```
using a ruby filter.
```
ruby {
code => '
data = event.get("jdbcOutput")
if data.is_a? Array
newData = []
data.each { |h|
q = h.delete("questionid")
newData << { "question#{q}" => h }
}
event.set("jdbcOutput", newData)
end
'
}
```
then split that.
---
### Author:  [@wwalker](https://discuss.elastic.co/u/wwalker)
#### Post date: [December 6, 2021, 8:45pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/6 "2021-12-06T20:45:42Z")
At least I was researching the right path, using ruby to manipulate the document. This is well and above anything I remotely understand with Ruby. Was I supposed to modify it, like the event.get from jdbcOutput to my actual field name, or just copy/paste? I copy/pasted it and nothing happens....no errors, no change in field structures.
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [December 6, 2021, 8:48pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/7 "2021-12-06T20:48:38Z")
You would need to update the event.get and .set calls to use the name of the field that your jdbc\_streaming filter fetched data into.
---
### Author:  [@wwalker](https://discuss.elastic.co/u/wwalker)
#### Post date: [December 6, 2021, 9:04pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/8 "2021-12-06T21:04:47Z")
Alright, that does the bulk of the work thank you. However, when splitting the fields out, they're all under `question` so the fields look like `question.question2.integeranswer` Is there a modification to this Ruby code to make it just `question2.integeranswer` or do I need to throw in an additional ruby script after this to modify them all?
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [December 6, 2021, 9:27pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/9 "2021-12-06T21:27:35Z")
Perhaps
```
data.each { |h|
q = h.delete("questionid")
event.set("question#{q}", h)
}
end
```
would work better for you, and maybe change the event.get to event.remove
Or if you need the array so that you can do the split then move the data using a ruby filter like [this one](https://discuss.elastic.co/t/how-to-dynamically-move-nested-key-value-to-root-level/180006/2).
---
### Author:  [@wwalker](https://discuss.elastic.co/u/wwalker)
#### Post date: [December 6, 2021, 9:36pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/10 "2021-12-06T21:36:56Z")
That did it, I appreciate the assistance. I've said it before, I need to find time to learn ruby so I can do this kind of stuff myself...or at the very least understand what the hell is going on here.
---
### Author:  [@wwalker](https://discuss.elastic.co/u/wwalker)
#### Post date: [December 7, 2021, 5:18pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/11 "2021-12-07T17:18:12Z")
Follow up question, can this ruby script be used to also remove the field if the value is null? I've tried using an additional ruby script I've used previously to remove empty fields, but it doesn't work on these fields, I'm assuming because they're nested.
```auto
ruby {
code => '
event.to_hash.each { |k, v|
if v.kind_of? String
if v == ""
event.remove(k)
end
end
}
'
}
```
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [December 7, 2021, 5:56pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/12 "2021-12-07T17:56:06Z")
> [@wwalker](#):
>
> I'm assuming because they're nested.
That is correct. That ruby filter only checks the top-level fields. [This](https://discuss.elastic.co/t/how-to-exclude-xml-json-key-value-if-key-length-is-greater-than-15-char-and-value-length-is-greater-than-100-char/270248/8) is an example of ruby code that recursively descends into fields of an event and modifies them.
---
### Author:  [@wwalker](https://discuss.elastic.co/u/wwalker)
#### Post date: [December 7, 2021, 9:48pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/13 "2021-12-07T21:48:50Z")
I'm very much lost in understanding most of this, but I THINK I changed it to remove null fields? Although, I think I'm missing something because I don't see where a target field is referencing...is it looking at the entire event or just the message field?
I'm not quite sure how to interpret what it's doing after the first couple logic checks. Define variable EmptyField, then if the object exists, and the object is a hash and not empty, then for each hash in the object do....something...
```auto
ruby {
code => '
def EmptyField(object, name, event)
if object
if object.kind_of?(Hash) and object != {}
object.each { |k, v| EmptyField(v, "#{name}[#{k}]", event) }
elsif object.kind_of?(Array) and object != []
object.each_index { |i|
EmptyField(object[i], "#{name}[#{i}]", event)
}
else
lastElement = name.gsub(/^.*\[/, "").gsub(/\]$/, "")
if lastElement.length = "null"
event.remove(name)
end
end
end
end
event.to_hash.each { |k, v|
EmptyField(v, "[#{k}]", event)
}
'
}
```
---
### Author:  [@wwalker](https://discuss.elastic.co/u/wwalker)
#### Post date: [December 7, 2021, 10:49pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/14 "2021-12-07T22:49:17Z")
No...this did not work...appears to remove everything in the event, lol.
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [December 7, 2021, 11:03pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/15 "2021-12-07T23:03:22Z")
> [@wwalker](#):
>
> `if lastElement.length = "null"`
That is never going to be true, so yes, it will delete everything. I think you can replace
```
lastElement = name.gsub(/^.*\[/, "").gsub(/\]$/, "")
if lastElement.length = "null"
```
with
```
if object = "null"
```
---
### Author:  [@wwalker](https://discuss.elastic.co/u/wwalker)
#### Post date: [December 7, 2021, 11:13pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/16 "2021-12-07T23:13:57Z")
I did catch that .length and removed it before testing it out. I made the change you suggested, but it still removes the whole event.
```auto
ruby {
code => '
def EmptyField(object, name, event)
if object
if object.kind_of?(Hash) and object != {}
object.each { |k, v| EmptyField(v, "#{name}[#{k}]", event) }
elsif object.kind_of?(Array) and object != []
object.each_index { |i|
EmptyField(object[i], "#{name}[#{i}]", event)
}
else
if object = "null"
event.remove(name)
end
end
end
end
event.to_hash.each { |k, v|
EmptyField(v, "[#{k}]", event)
}
'
}
```
I'm running this AFTER the first ruby script that restructures the object to below. Should it be running first?
```auto
"question": {
"question2": {
"textanswer": null
},
"question4": {
"textanswer": null
}
}
```
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [December 7, 2021, 11:39pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/17 "2021-12-07T23:39:30Z")
> [@wwalker](#):
>
> `if object = "null"`
Sorry, that is an assignment so it is unconditionally true. Try `if object == "null"`. Note that null in JSON will be nil in Ruby, so you may need `if ! object` instead.
I would run it after the restructuring.
---
### Author:  [@wwalker](https://discuss.elastic.co/u/wwalker)
#### Post date: [December 7, 2021, 11:46pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/18 "2021-12-07T23:46:08Z")
Tried `if object == "null"`, `if object == "nil"`, and `if ! object` and none of them worked. They didn't delete the event, but they didn't seem to do anything either.
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [December 7, 2021, 11:52pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/19 "2021-12-07T23:52:03Z")
What does an event look like if you use
```
output { stdout { codec => rubydebug } }
```
(Doesn't have to be stdout, you could use a file output if it is more convenient.)
---
### Author:  [@wwalker](https://discuss.elastic.co/u/wwalker)
#### Post date: [December 8, 2021, 3:32pm UTC](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012/20 "2021-12-08T15:32:59Z")
```auto
{
"question39": {
"dec": null,
"question": "Sup Brah?",
"bool": null,
"date": null,
"text": null,
"int": null
}
}
```
[Next page](https://discuss.elastic.co/t/processing-multiple-rows-from-jdbc-streaming/291012.md?page=2)