How to convert a CSV file into a JSON, line by line, using LUA?

[rcmv]

Hi!

I would like to convert a CSV file into JSON line by line. I think I can use the remap function with parse_json for this effect but I'm very noob with this and sure how to make it. My attempts failed.

Can someone help me with this simple example of a CSV below?

Timestamp;country;duration;quality
06/09/2022 10:33;UAE;7.8470;3.5362
06/09/2022 10:30;Saudi_Arabia;10.0380;3.4051
06/09/2022 10:27;Switzerland;7.8720;2.4598
06/09/2022 10:24;Israel;9.9360;2.8292

Edit:
After some search in the documentation I tried to implement this using Lua with the code below:

`[sources.my_csv_files_id]
type = "file"
include = [ "/home/rcmv2/ftp/" ]
start_at_beginning = true

[transforms.lua]
inputs = ["my_csv_files_id"]
type = "lua"
version = "2"
source = """
    csv = require("csv") 
    column_names = {  
        "Timestamp",
	"country",
	"duration",
	"quality",
    }
"""
hooks.process = """
  function (event, emit)
    fields = csv.openstring(event.log.message):lines()() 
    event.log.message = nil 

    for column, value in ipairs(fields) do 
      column_name = column_names[column] 
      event.log[column_name] = value 
    end

    emit(event) 
  end
"""
[sinks.console]
inputs = ["my_csv_files_id"]
type = "console"
encoding.codec = "json"`

I'm not get any output though...
Any help?
Thanks

[jszwedko]

Hi @rcmv ,

I think your issue is that your transform is being skipped as you've specified the source, my_csv_files_id, as the input to the sink rather than lua.

However, I'd also recommend looking at remap and parse_csv. Something like:

[sources.my_csv_files_id]
type = "file"
include = [ "/home/rcmv2/ftp/" ]
start_at_beginning = true

[transforms.remap]
inputs = ["my_csv_files_id"]
type = "remap"
source = """
    csv = require("csv") 
    row = parse_csv!(.message)
    .Timestamp = row[0]
    .country = row[1]
   .duration = row[2]
   .quality = row[3]
"""

[sinks.console]
inputs = ["remap"]
type = "console"
encoding.codec = "json"

I didn't test it, but something like that should work.

[rcmv]

Hi @jszwedko
Thanks for the reply. I've changed the code as per your suggestion but the behavior is the same.
Code:

[sources.my_csv_files_id]
type = "file"
include = [ "/home/rcmv2/ftp/" ]
start_at_beginning = true

[transforms.lua]
inputs = ["my_csv_files_id"]
type = "lua"
version = "2"
source = """
    csv = require("csv") -- load external module for parsing CSV
    column_names = {  -- a sequence containing CSV column names
      "Timestamp",
          "country",
          "duration",
          "quality",
    }
"""
hooks.process = """
  function (event, emit)
    fields = csv.openstring(event.log.message):lines()() -- parse the `message` field
    event.log.message = nil -- drop the `message` field

    for column, value in ipairs(fields) do -- iterate over CSV columns
      column_name = column_names[column] -- get column name
      event.log[column_name] = value -- set the corresponding field in the event
    end

    emit(event) -- emit the transformed event
  end
"""
[sinks.console]
inputs = ["lua"]
type = "console"
encoding.codec = "json"

Output:

2022-09-07T08:43:31.787580Z  INFO vector::topology::builder: Healthcheck: Passed.
2022-09-07T08:43:31.787804Z  INFO source{component_kind="source" component_id=my_csv_files_id component_type=file component_name=my_csv_files_id}: vector::sources::file: Starting file server. include=["/home/rcmv2/ftp/"] exclude=[]
2022-09-07T08:43:31.788166Z  INFO source{component_kind="source" component_id=my_csv_files_id component_type=file component_name=my_csv_files_id}:file_server: file_source::checkpointer: Loaded checkpoint data.


2022-09-07T08:44:40.516737Z  INFO vector: Vector has stopped.

I'll have a look on your remap/parse_csv approach.
Thanks!

[rcmv]

Hi again @jszwedko !
I've also tested the remap/parse_csv approach but I'm basically getting the same behavior. I had to change the code a bit because "csv = require("csv")" gives me an error message indicating it is an undefined function so I removed it. With the code below I dont get any error message but I also dont get any relevant output.
Code:

[sources.my_csv_files_id]
type = "file"
include = [ "/home/rcmv2/ftp/" ]
start_at_beginning = true

[transforms.remap]
inputs = ["my_csv_files_id"]
type = "remap"
source = """
     row = parse_csv!(.message)
    .Timestamp = row[0]
    .country = row[1]
   .duration = row[2]
   .quality = row[3]
"""

[sinks.console]
inputs = ["remap"]
type = "console"
encoding.codec = "json"

Output:

2022-09-07T08:54:40.260166Z  INFO vector::app: Log level is enabled. level="vector=info,codec=info,vrl=info,file_source=info,tower_limit=trace,rdkafka=info,buffers=info,kube=info"
2022-09-07T08:54:40.260378Z  INFO vector::app: Loading configs. paths=["vector.toml"]
2022-09-07T08:54:40.264096Z  WARN vector::sources::file: Use of deprecated option `start_at_beginning`. Please use `ignore_checkpoints` and `read_from` options instead.
2022-09-07T08:54:40.265713Z  INFO vector::topology::running: Running healthchecks.
2022-09-07T08:54:40.265872Z  INFO vector::topology::builder: Healthcheck: Passed.
2022-09-07T08:54:40.266079Z  INFO source{component_kind="source" component_id=my_csv_files_id component_type=file component_name=my_csv_files_id}: vector::sources::file: Starting file server. include=["/home/rcmv2/ftp/"] exclude=[]
2022-09-07T08:54:40.266466Z  INFO vector: Vector has started. debug="false" version="0.24.0" arch="x86_64" build_id="43267b9 2022-08-30"
2022-09-07T08:54:40.266598Z  INFO vector::app: API is disabled, enable by setting `api.enabled` to `true` and use commands like `vector top`.
2022-09-07T08:54:40.266821Z  INFO source{component_kind="source" component_id=my_csv_files_id component_type=file component_name=my_csv_files_id}:file_server: file_source::checkpointer: Loaded checkpoint data.
2022-09-07T08:55:47.922535Z  INFO vector: Vector has stopped.
2022-09-07T08:55:47.922880Z  INFO vector::topology::running: Shutting down... Waiting on running components. remaining_components="my_csv_files_id, console, remap" time_remaining="59 seconds left"

[rcmv]

I got the bug using the Remap/parse_csv approach. With Lua still does not work. The problem was the csv file delimiter. The delimiter was ';' and I think the parse_csv use by default the delimiter ','.
I changed my csv file delimiter from ; to , and now I'm getting the following output:

{"Timestamp":"Timestamp","country":"country","duration":"duration","file":"/home/rcmv2/ftp/sample.csv","host":"ip-172-31-28-49.ec2.internal","message":"Timestamp,country,duration,quality\r","quality":"quality","source_type":"file","timestamp":"2022-09-07T13:52:18.056567562Z"} {"Timestamp":"06/09/2022 10:33","country":"UAE","duration":"7.8470","file":"/home/rcmv2/ftp/sample.csv","host":"ip-172-31-28-49.ec2.internal","message":"06/09/2022 10:33,UAE,7.8470,3.5362\r","quality":"3.5362","source_type":"file","timestamp":"2022-09-07T13:52:18.056578071Z"} {"Timestamp":"06/09/2022 10:30","country":"Saudi_Arabia","duration":"10.0380","file":"/home/rcmv2/ftp/sample.csv","host":"ip-172-31-28-49.ec2.internal","message":"06/09/2022 10:30,Saudi_Arabia,10.0380,3.4051\r","quality":"3.4051","source_type":"file","timestamp":"2022-09-07T13:52:18.056583668Z"} {"Timestamp":"06/09/2022 10:27","country":"Switzerland","duration":"7.8720","file":"/home/rcmv2/ftp/sample.csv","host":"ip-172-31-28-49.ec2.internal","message":"06/09/2022 10:27,Switzerland,7.8720,2.4598\r","quality":"2.4598","source_type":"file","timestamp":"2022-09-07T13:52:18.056588890Z"} {"Timestamp":"06/09/2022 10:24","country":"Israel","duration":"9.9360","file":"/home/rcmv2/ftp/sample.csv","host":"ip-172-31-28-49.ec2.internal","message":"06/09/2022 10:24,Israel,9.9360,2.8292\r","quality":"2.8292","source_type":"file","timestamp":"2022-09-07T13:52:18.056594076Z"}
Not sure what is the syntax to change the delimiter using parse_csv . I tried row = parse_csv!(.message, delimiter=';') but it didn't worked.