# Logstash Plugins: input-webhdfs
**URL:** https://discuss.elastic.co/t/logstash-plugins-input-webhdfs/80656
**Category:** Logstash
**Created:** [March 30, 2017, 10:55am UTC](https://discuss.elastic.co/t/logstash-plugins-input-webhdfs/80656 "2017-03-30T10:55:43Z")
**Posts on this page:** 4
**Page:** 1
### Author:  [@ohk](https://discuss.elastic.co/u/ohk)
#### Post date: [March 30, 2017, 10:55am UTC](https://discuss.elastic.co/t/logstash-plugins-input-webhdfs/80656/1 "2017-03-30T10:55:43Z")
Hi,
I need a plugin able to read logs files from HDFS. I took the output-webhdfs plugin and changed to download files from HDFS using its REST API then parsing this temporary files with adequate codec.
For now I can connect to hdfs and get the files, but I have some troubles when using json\_lines codec to parse them. I will share my code, I need you're help to figure out what I'm doing wrong and eventually if you have some recommendations. [code here](https://drive.google.com/open?id=0B-1k-3U7NiNpdUNMSHRvRDUxTFU)
Thank you in advance
---
### Author:  [@pierhugues](https://discuss.elastic.co/u/pierhugues)
#### Post date: [March 31, 2017, 4:34pm UTC](https://discuss.elastic.co/t/logstash-plugins-input-webhdfs/80656/2 "2017-03-31T16:34:56Z")
Hello @ohk,
What are exactly the symptoms you are having?
- Incorrectly parsed data?
- Logstash hang?
- Any errors in the log?
What is the structure of your document that you are fetching from HDFS? It is one JSON document per line, In this case the _json_ codec might be more appropriate and you make your input code extract one line at a time and send it to the decode method of the codec.
I've taken a quick look at the code, I don't know the details of the library but I can give you this recommendation.
I think there is a small problem in the `#parse_events` method, you are trying to read from the client `@client.read`, write to a file, read from the file that you are currently writing?
```ruby
def parse_events(filename, out_queue)
FileUtils::mkdir_p "/tmp/logstash-input-webhdfs"
File.open("/tmp/logstash-input-webhdfs/tmp_#{filename.to_i(32)}", 'w+') do |io|
io.write(@client.read(filename))
@codec.decode(io.read) do |event|
event["host"] = @host+":"+@port
event["filename"] = filename
decorate(event)
@logger.debug("Event : #{event}")
out_queue << event
end
end
end # def parse_events
```
I don't know what `@client.read` returns, it is line or chunk of text? I think you could solve this problem in two differents ways:
1.
- Read the file from the service, write it to disk
- Process the file, read the file from the beginning and process each line.
1.
- If you can get line with `client#read`, you just send line directly to the code.
---
### Author:  [@ohk](https://discuss.elastic.co/u/ohk)
#### Post date: [April 3, 2017, 3:04pm UTC](https://discuss.elastic.co/t/logstash-plugins-input-webhdfs/80656/3 "2017-04-03T15:04:33Z")
Thank you very much for your help. I changed codec to json and I follower the first approach (write to file then read file line by line) and it works. But I still have a warning when inserting to elastic.
my input files look like this: [input file](https://drive.google.com/open?id=0B-1k-3U7NiNpbDhRNVlvWm0yNnM)
Warning message:
> [2017-04-03T16:36:41,232][WARN][logstash.outputs.elasticsearch] Failed action. {:status=\>400, :action=\>["index", {:\_id=\>nil, :\_index=\>"logstash-2017.04.03", :\_type=\>"logs", :\_routing=\>nil}, 2017-04-03T14:36:41.065Z %{host} %{message}], :response=\>{"index"=\>{"\_index"=\>"logstash-2017.04.03", "\_type"=\>"logs", "\_id"=\>"AVs0PxVnWIlWG4bx6UaU", "status"=\>400, "error"=\>{"type"=\>"illegal\_argument\_exception", "reason"=\>"Can't merge a non object mapping [Properties.spark.rdd.scope] with an object mapping [Properties.spark.rdd.scope]"}}}}
> [2017-04-03T16:36:41,232][WARN][logstash.outputs.elasticsearch] Failed action. {:status=\>400, :action=\>["index", {:\_id=\>nil, :\_index=\>"logstash-2017.04.03", :\_type=\>"logs", :\_routing=\>nil}, 2017-04-03T14:36:41.066Z %{host} %{message}], :response=\>{"index"=\>{"\_index"=\>"logstash-2017.04.03", "\_type"=\>"logs", "\_id"=\>"AVs0PxVnWIlWG4bx6UaV", "status"=\>400, "error"=\>{"type"=\>"illegal\_argument\_exception", "reason"=\>"Can't merge a non object mapping [Properties.spark.rdd.scope] with an object mapping [Properties.spark.rdd.scope]"}}}}
I think it's linked to the mapping in elastic.
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [May 1, 2017, 3:04pm UTC](https://discuss.elastic.co/t/logstash-plugins-input-webhdfs/80656/4 "2017-05-01T15:04:33Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.