I am trying to fetch data from a sparql endpoint.
i can make a successfull http call with the following parameters:
endpoint: cr.eionet.europa.eu/sparql
header set like:
"query"= "see the query in logstash.conf...easier"
"Content-type"= "application/sparql-query"
"format"= "text/csv"
but my logstash.conf fails miserably:
input {
http_poller {
urls => {
data => {
method => post
url => "${ENDPOINT}"
headers => {
Content-type => "application/sparql-query"
format => "text/csv"
query => "PREFIX rdfns: http://www.w3.org/1999/02/22-rdf-syntax-ns# PREFIX rod: http://rod.eionet.europa.eu/schema.rdf# PREFIX rdfsh: http://www.w3.org/2000/01/rdf-schema# PREFIX dc: http://purl.org/dc/terms/ PREFIX cr: http://cr.eionet.europa.eu/ontologies/contreg.rdf# PREFIX ns: http://www.w3.org/ns/dcat# PREFIX skos: http://www.w3.org/2004/02/skos/core# PREFIX contreg: http://cr.eionet.europa.eu/ontologies/contreg.rdf# SELECT DISTINCT md5(sql:GROUP_CONCAT(?subj, ?fileUri )) as ?_id ?subj as ?envelopeUri ?obligationUri ?obligationTitle ?instrumentUri ?instrumentTitle ?iso2 ?country ?envelopeTitle ?envelopeReporter ?firstSeen ?released ?feedbackUri ?feedbackTitle ?FeedbackStatus ?fileUri ?filename xsd:double(?filebyteSize)/1000.0 as ?fileKbSize ?fileMediaType WHERE {{ SELECT DISTINCT * WHERE { ?subj a rod:Delivery . ?subj rod:locality ?locality . ?subj rod:obligation ?obligationUri . ?subj contreg:firstSeen ?firstSeen . ?subj rod:released ?released . ?subj dc:title ?envelopeTitle . ?subj dc:creator ?envelopeReporter . ?subj rod:hasFile ?fileUri. ?fileUri rdfsh:label ?filename . ?fileUri ns:byteSize ?filebyteSize . ?fileUri cr:mediaType ?fileMediaType . ?locality rod:localityName ?country . ?locality skos:notation ?iso2 . ?obligationUri dc:title ?obligationTitle . ?obligationUri rod:instrument ?instrumentUri . ?instrumentUri rdfs:label ?instrumentTitle . OPTIONAL { ?subj cr:hasFeedback ?feedbackUri . ?feedbackUri dc:title ?feedbackTitle . ?feedbackUri cr:feedbackStatus ?FeedbackStatus . ?feedbackUri cr:feedbackFor ?fileUri . } FILTER NOT EXISTS {?feedbackUri cr:feedbackFor ?fileUri} } } UNION { SELECT DISTINCT * WHERE { ?subj a rod:Delivery . ?subj rod:locality ?locality . ?subj rod:obligation ?obligationUri . ?subj contreg:firstSeen ?firstSeen . ?subj rod:released ?released . ?subj dc:title ?envelopeTitle . ?subj dc:creator ?envelopeReporter . ?locality rod:localityName ?country . ?locality skos:notation ?iso2 . ?obligationUri dc:title ?obligationTitle . ?obligationUri rod:instrument ?instrumentUri . ?instrumentUri rdfs:label ?instrumentTitle . ?subj cr:hasFeedback ?feedbackUri . ?feedbackUri dc:title ?feedbackTitle . ?feedbackUri cr:feedbackStatus ?FeedbackStatus . OPTIONAL { ?feedbackUri cr:feedbackFor ?fileUri . ?fileUri rdfsh:label ?filename . ?fileUri ns:byteSize ?filebyteSize . ?fileUri cr:mediaType ?fileMediaType . } } } FILTER (year(?released) = 2017) }"
}
}
}
request_timeout => 60
# Supports "cron", "every", "at" and "in" schedules by rufus scheduler
schedule => { cron => "* * * * * UTC"}
codec => "json"
# A hash of request metadata info (timing, response headers, etc.) will be sent here
metadata_target => "http_poller_metadata"
}
}
filter {
csv {
separator => ";"
columns => [
"_id",
"envelopeUri",
"obligationUri",
"obligationTitle",
"instrumentUri",
"instrumentTitle",
"iso2",
"country",
"envelopeTitle",
"envelopeReporter",
"firstSeen",
"released",
"feedbackUri",
"feedbackTitle",
"FeedbackStatus",
"fileUri",
"filename",
"fileKbSize",
"fileMediaType",
"year"
]
convert => {
'year' => 'integer'
'fileKbSize' => 'integer'
'released' => 'date'
}
}
mutate { }
}
}
output {
elasticsearch {
hosts => "elasticsearch:9200"
index => "${INDEXNAME}"
document_id => "%{_id}"
user => "${LOGSTASH_RW_USERNAME}"
password => "${LOGSTASH_RW_PASSWORD}"
ssl => true
ssl_certificate_verification => false
}
stdout {}
}
my output is:
Sending Logstash's logs to /var/log/logstash which is now configured via log4j2.properties
15:13:40.385 [main] INFO logstash.modules.scaffold - Initializing module {:module_name=>"fb_apache", :directory=>"/usr/share/logstash/modules/fb_apache/configuration"}
15:13:40.390 [main] INFO logstash.modules.scaffold - Initializing module {:module_name=>"netflow", :directory=>"/usr/share/logstash/modules/netflow/configuration"}
15:13:40.552 [LogStash::Runner] ERROR logstash.agent - Cannot create pipeline {:reason=>"Expected one of #, => at line 8, column 18 (byte 145) after input {\n http_poller {\n urls => {\n data => {\n method => post\n url => "${ENDPOINT}"\n headers => {\n Content"}
basically logstash doesn't like my conf file from the very beginning. What am I missing?