How to convert multiple CSVs, with different number of columns, in one shot, into JSON?

[rcmv]

Hi!

I have three different CSV files. Below the CSVs (with only one row for simplicity sake):
CSV Type 1:

Timestamp,a_country,b_country,call_setup_time,test_type
06/09/2022 10:33,UAE,UAE,7.8470,voice_call

CSV Type 2:

Timestamp,a_country,b_country,call_setup_time,quality,test_type
05/09/2022 12:33,Portugal,UK,4.8470,3.9362,voice quality

CSV Type 3:

Timestamp,a_country,latency,throughput,test type
05/09/2022 12:33,Portugal,40.8470,20.1,data

Now I'm in doubt on how to do this.
I guess I would need to collect first, the columns of the CSV being processed (missing code on the first if statement) and based on those columns, fill the associated values.
Sample code below:

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

[transforms.remap]
inputs = ["my_csv_files_id"]
type = "remap"
source = """
    row = parse_csv!(.message)
    if row[0] == "Timestamp" {
        # code to collect the headers of the CSV type being processed
    }else{
	# code to, based on the headers collected before, get the associated data.
    }
    del(.source_type)
    del(.timestamp)
    del(.file)
    del(.host)
    del(.message)
   }
    """
[sinks.console]
inputs = ["remap"]
type = "console"
encoding.codec = "json"

Desired Output:

{"Timestamp":"06/09/2022 10:33","a_country":"UAE";"b_country":"UAE";"call_setup_time":"7.8470","test_type":"voice_call"}

{"Timestamp":"05/09/2022 12:33","a_country":"Portugal","b_country":"UK","call_setup_time":"4.8470","quality":"3.9362","test_type":"voice quality"}

{"Timestamp":"05/09/2022 12:33","a_country":"Portugal","latency":"40.8470","throughput":"20.1","test type":"data"}

Any ideas about the code inside those two if statements?

Thanks

[jszwedko]

Hi @rcmv !

You could probably use lua for this by caching the headers for each file by the name of the file (in a Lua table) since that transform is stateful and will persist state across events. remap is stateless so there wouldn't be a way to hold onto the headers.

I'd suggest you consider changing your input files or preprocessing them before having Vector ingest them, though, to a format more easily read by Vector (e.g. JSON). That is: convert the CSV files to JSON and then have Vector read those.

See https://vector.dev/guides/advanced/parsing-csv-logs-with-lua/ for a guide for parsing CSV with Lua if you do decide to pursue that route.

[rcmv]

Hi @jszwedko!
I follow your recommendation and I'm preprocessing the csv files with python. I merged the CSVs and converted them into one json file.
I've two options regarding the json output. On the first output version I've something like this:

[
{"Timestamp": "06/09/2022 10:33", "a_country": "UAE", "b_country": "UAE", "call_setup_time": "7.847", "quality": "", "latency": "", "throughput": "", "test_type": "voice_call"},
 {"Timestamp": "06/09/2022 10:30", "a_country": "Saudi_Arabia", "b_country": "Saudi_Arabia", "call_setup_time": "10.038", "quality": "", "latency": "", "throughput": "", "test_type": "voice_call"},
 {"Timestamp": "06/09/2022 10:27", "a_country": "Switzerland", "b_country": "Switzerland", "call_setup_time": "7.872", "quality": "", "latency": "", "throughput": "", "test_type": "voice_call"},
 {"Timestamp": "06/09/2022 10:24", "a_country": "Israel", "b_country": "Israel", "call_setup_time": "9.936", "quality": "", "latency": "", "throughput": "", "test_type": "voice_call"},
 {"Timestamp": "05/09/2022 12:33", "a_country": "Portugal", "b_country": "UK", "call_setup_time": "4.847", "quality": "3.9362", "latency": "", "throughput": "", "test_type": "voice quality"},
 {"Timestamp": "05/09/2022 12:30", "a_country": "Espanha", "b_country": "USA", "call_setup_time": "5.038", "quality": "3.1051", "latency": "", "throughput": "", "test_type": "voice quality"},
 {"Timestamp": "05/09/2022 12:27", "a_country": "UK", "b_country": "Italy", "call_setup_time": "9.872", "quality": "3.4598", "latency": "", "throughput": "", "test_type": "voice quality"},
 {"Timestamp": "05/09/2022 12:24", "a_country": "USA", "b_country": "Croatia", "call_setup_time": "8.936", "quality": "1.8292", "latency": "", "throughput": "", "test_type": "voice quality"},
 {"Timestamp": "05/09/2022 12:33", "a_country": "Portugal", "b_country": "", "call_setup_time": "", "quality": "", "latency": "40.847", "throughput": "20.1", "test_type": "data"},
 {"Timestamp": "05/09/2022 12:30", "a_country": "Espanha", "b_country": "", "call_setup_time": "", "quality": "", "latency": "50.038", "throughput": "35.67", "test_type": "data"},
 {"Timestamp": "05/09/2022 12:27", "a_country": "UK", "b_country": "", "call_setup_time": "", "quality": "", "latency": "90.872", "throughput": "3.4598", "test_type": "data"},
 {"Timestamp": "05/09/2022 12:24", "a_country": "USA", "b_country": "", "call_setup_time": "", "quality": "", "latency": "80.936", "throughput": "1.8292", "test_type": "data"}
 ]

The second output version is not a valid json file format but may be easier to ingest with vector. It basically has each row of the csv in json:

{"Timestamp": "06/09/2022 10:33", "a_country": "UAE", "b_country": "UAE", "call_setup_time": "7.847", "quality": "", "latency": "", "throughput": "", "test_type": "voice_call"}
{"Timestamp": "06/09/2022 10:30", "a_country": "Saudi_Arabia", "b_country": "Saudi_Arabia", "call_setup_time": "10.038", "quality": "", "latency": "", "throughput": "", "test_type": "voice_call"}{"Timestamp": "06/09/2022 10:27", "a_country": "Switzerland", "b_country": "Switzerland", "call_setup_time": "7.872", "quality": "", "latency": "", "throughput": "", "test_type": "voice_call"}{"Timestamp": "06/09/2022 10:24", "a_country": "Israel", "b_country": "Israel", "call_setup_time": "9.936", "quality": "", "latency": "", "throughput": "", "test_type": "voice_call"}{"Timestamp": "05/09/2022 12:33", "a_country": "Portugal", "b_country": "UK", "call_setup_time": "4.847", "quality": "3.9362", "latency": "", "throughput": "", "test_type": "voice quality"}{"Timestamp": "05/09/2022 12:30", "a_country": "Espanha", "b_country": "USA", "call_setup_time": "5.038", "quality": "3.1051", "latency": "", "throughput": "", "test_type": "voice quality"}{"Timestamp": "05/09/2022 12:27", "a_country": "UK", "b_country": "Italy", "call_setup_time": "9.872", "quality": "3.4598", "latency": "", "throughput": "", "test_type": "voice quality"}{"Timestamp": "05/09/2022 12:24", "a_country": "USA", "b_country": "Croatia", "call_setup_time": "8.936", "quality": "1.8292", "latency": "", "throughput": "", "test_type": "voice quality"}{"Timestamp": "05/09/2022 12:33", "a_country": "Portugal", "b_country": "", "call_setup_time": "", "quality": "", "latency": "40.847", "throughput": "20.1", "test_type": "data"}{"Timestamp": "05/09/2022 12:30", "a_country": "Espanha", "b_country": "", "call_setup_time": "", "quality": "", "latency": "50.038", "throughput": "35.67", "test_type": "data"}{"Timestamp": "05/09/2022 12:27", "a_country": "UK", "b_country": "", "call_setup_time": "", "quality": "", "latency": "90.872", "throughput": "3.4598", "test_type": "data"}{"Timestamp": "05/09/2022 12:24", "a_country": "USA", "b_country": "", "call_setup_time": "", "quality": "", "latency": "80.936", "throughput": "1.8292", "test_type": "data"}

Now, I'm changing the vector config file to ingest these two files (one at a time, in the end I will only need one version but for now I'm trying to see which one is more easily ingested) to be able to sink each row in a json format, but I'm getting the following warning (for both files):

2022-09-09T14:02:01.720055Z  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/*.json"] exclude=[]
2022-09-09T14:02:01.720460Z  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-09T14:02:01.720709Z  WARN source{component_kind="source" component_id=my_csv_files_id component_type=file component_name=my_csv_files_id}:file_server: vector::internal_events::file::source: Currently ignoring file too small to fingerprint. file=/home/rcmv2/ftp/global_file.json

Below my vector config (now completely straight forward):

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

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

Any idea on why this is happening?

[jszwedko]

Hi @rcmv !

Great! Newlined-delimited JSON, as you have in your second, looks good and should be readable by the file source. It'll be harder to read the first example which has a JSON array.

For:

2022-09-09T14:02:01.720709Z  WARN source{component_kind="source" component_id=my_csv_files_id component_type=file component_name=my_csv_files_id}:file_server: vector::internal_events::file::source: Currently ignoring file too small to fingerprint. file=/home/rcmv2/ftp/global_file.json

To create a unique fingerprint for a file, Vector will use the first line. That message indicates that the referenced file doesn't have more than one line in it. Is that accurate?

[rcmv]

Hi @jszwedko !
Sorry for the delay. Weekend got in the middle. You were right once again. I changed my python script to add a newline character after each '}' and now the file has each 'row' on a different line. On my first implementation of the python script I add all the text on the first line (I've manually changed it in my previous post here to make it easy to read). I corrected this and now the output of my vector config is:

`{"message":{"Timestamp":"06/09/2022 10:33","a_country":"UAE","b_country":"UAE","call_setup_time":"7.847","latency":"","quality":"","test_type":"voice_call","throughput":""}}
{"message":{"Timestamp":"06/09/2022 10:30","a_country":"Saudi_Arabia","b_country":"Saudi_Arabia","call_setup_time":"10.038","latency":"","quality":"","test_type":"voice_call","throughput":""}}
{"message":{"Timestamp":"06/09/2022 10:27","a_country":"Switzerland","b_country":"Switzerland","call_setup_time":"7.872","latency":"","quality":"","test_type":"voice_call","throughput":""}}
{"message":{"Timestamp":"06/09/2022 10:24","a_country":"Israel","b_country":"Israel","call_setup_time":"9.936","latency":"","quality":"","test_type":"voice_call","throughput":""}}
{"message":{"Timestamp":"05/09/2022 12:33","a_country":"Portugal","b_country":"UK","call_setup_time":"4.847","latency":"","quality":"3.9362","test_type":"voice quality","throughput":""}}
{"message":{"Timestamp":"05/09/2022 12:30","a_country":"Espanha","b_country":"USA","call_setup_time":"5.038","latency":"","quality":"3.1051","test_type":"voice quality","throughput":""}}
{"message":{"Timestamp":"05/09/2022 12:27","a_country":"UK","b_country":"Italy","call_setup_time":"9.872","latency":"","quality":"3.4598","test_type":"voice quality","throughput":""}}
{"message":{"Timestamp":"05/09/2022 12:24","a_country":"USA","b_country":"Croatia","call_setup_time":"8.936","latency":"","quality":"1.8292","test_type":"voice quality","throughput":""}}
{"message":{"Timestamp":"05/09/2022 12:33","a_country":"Portugal","b_country":"","call_setup_time":"","latency":"40.847","quality":"","test_type":"data","throughput":"20.1"}}
{"message":{"Timestamp":"05/09/2022 12:30","a_country":"Espanha","b_country":"","call_setup_time":"","latency":"50.038","quality":"","test_type":"data","throughput":"35.67"}}
{"message":{"Timestamp":"05/09/2022 12:27","a_country":"UK","b_country":"","call_setup_time":"","latency":"90.872","quality":"","test_type":"data","throughput":"3.4598"}}
`

I've now two problems.
1º: By some reason, vector is changing the column order of the JSON file in the output above. Is there any way to prevent this?
Below you can see the (correct) column order of the JSON file being ingested:

`cat global_file_w.json

{"Timestamp": "06/09/2022 10:33", "a_country": "UAE", "b_country": "UAE", "call_setup_time": "7.847", "quality": "", "latency": "", "throughput": "", "test_type": "voice_call"}
{"Timestamp": "06/09/2022 10:30", "a_country": "Saudi_Arabia", "b_country": "Saudi_Arabia", "call_setup_time": "10.038", "quality": "", "latency": "", "throughput": "", "test_type": "voice_call"}
{"Timestamp": "06/09/2022 10:27", "a_country": "Switzerland", "b_country": "Switzerland", "call_setup_time": "7.872", "quality": "", "latency": "", "throughput": "", "test_type": "voice_call"}
{"Timestamp": "06/09/2022 10:24", "a_country": "Israel", "b_country": "Israel", "call_setup_time": "9.936", "quality": "", "latency": "", "throughput": "", "test_type": "voice_call"}
{"Timestamp": "05/09/2022 12:33", "a_country": "Portugal", "b_country": "UK", "call_setup_time": "4.847", "quality": "3.9362", "latency": "", "throughput": "", "test_type": "voice quality"}
{"Timestamp": "05/09/2022 12:30", "a_country": "Espanha", "b_country": "USA", "call_setup_time": "5.038", "quality": "3.1051", "latency": "", "throughput": "", "test_type": "voice quality"}
{"Timestamp": "05/09/2022 12:27", "a_country": "UK", "b_country": "Italy", "call_setup_time": "9.872", "quality": "3.4598", "latency": "", "throughput": "", "test_type": "voice quality"}
{"Timestamp": "05/09/2022 12:24", "a_country": "USA", "b_country": "Croatia", "call_setup_time": "8.936", "quality": "1.8292", "latency": "", "throughput": "", "test_type": "voice quality"}
{"Timestamp": "05/09/2022 12:33", "a_country": "Portugal", "b_country": "", "call_setup_time": "", "quality": "", "latency": "40.847", "throughput": "20.1", "test_type": "data"}
{"Timestamp": "05/09/2022 12:30", "a_country": "Espanha", "b_country": "", "call_setup_time": "", "quality": "", "latency": "50.038", "throughput": "35.67", "test_type": "data"}
{"Timestamp": "05/09/2022 12:27", "a_country": "UK", "b_country": "", "call_setup_time": "", "quality": "", "latency": "90.872", "throughput": "3.4598", "test_type": "data"}
{"Timestamp": "05/09/2022 12:24", "a_country": "USA", "b_country": "", "call_setup_time": "", "quality": "", "latency": "80.936", "throughput": "1.8292", "test_type": "data"}`

2º: Is there a more elegant way to change the vector output in order to remove the 'message' string without having to specify each column name in the vector config file? I mean, I just want to remove the 'message' part. Example:
From this:
{"message":{"Timestamp":"05/09/2022 12:30","a_country":"Espanha","b_country":"","call_setup_time":"","latency":"50.038","quality":"","test_type":"data","throughput":"35.67"}}
To this:
{"Timestamp":"05/09/2022 12:30","a_country":"Espanha","b_country":"","call_setup_time":"","latency":"50.038","quality":"","test_type":"data","throughput":"35.67"}

Thanks for all the support!

[jszwedko]

Glad you got it working!

1º: By some reason, vector is changing the column order of the JSON file in the output above. Is there any way to prevent this?

Unfortunately not. Vector alphabetizes objects/maps when serializing them. We are open to a feature request here though.

2º: Is there a more elegant way to change the vector output in order to remove the 'message' string without having to specify each column name in the vector config file? I mean, I just want to remove the 'message' part.

You should be able to do:

. = .message

in a remap transform to bring all of the fields up.

[rcmv]

Thank you @jszwedko !
Issue solved :)