How do you process an entire CSV file all at one time

[Manniac07]

The complete csv file arrives in source folder and I am using lua to load the entire csv into an array then I emit each cell that matches of each line that matches my criteria until EOF. The issue is that vector reprocesses the source csv file for every row in the csv file.

I have attempted to use multiline from BOF to EOF and it still processes line by line.

I have also tried to remove file using remove_after_secs and that didn't work either.

[jszwedko]

Hi @Manniac07 ,

I'm not sure I completely understand. Can you share your configuration?

[Manniac07]

yes @jszwedko
source:

`
  data_dir = "/var/lib/vector"
# Random Syslog-formatted logs
[sources.vectorsiprealmmethod]
 type = "file"
 fingerprint.lines = 5
 fingerprint.strategy = "checksum"
 fingerprint.ignored_header_bytes = 0
 glob_minimum_cooldown_ms = 10
 include = ["/home/abc/**/sip-realm-method/*.csv"]
 oldest_first = true
 read_from = "beginning"

  [sources.vectorsiprealmmethod.multiline]
    start_pattern = '\\\A'             # absolute string start
    mode = "halt_with"
    condition_pattern = '\\\z'          # absolute string end
    timeout_ms = 1000
`

transform

`
[transforms.vectorlua]
 type = "lua"
 inputs = [ "vectorsiprealmmethod" ]
 version = "2"
 source = """
function init(event)
    csv = require("csv")
    valuesArray = {}
    rowNum = 1
    rowIndx = -1
    colIndx = -1
    processedPath = ""
    path = ""
    pathName = ""
    pathserver = ""
    datacenter = ""
    servertype = ""
end
function setFileParms(path)                                      
    path = string.gsub(path, "-", "_")
    if path ~=nil and string.find(path,"sip_realm_method") then	
	if path ~=nil and string.find(path,"SAT_SIP_CU1_TRPSBC1_SAT_SIP_CU1_TRPSBC2")   then
            pathName = "vis_sip_realm_method"
            pathserver = "SAT-SIP-CU1-TRPSBC1_SAT-SIP-CU1-TRPSBC2"
		    datacenter = "SAT"
		    servertype = "Transport"
	elseif path ~=nil and string.find(path,"DFW_SIP_DALEQ_TEST_TRPSBC1") then
            pathName = "vis_sip_realm_method"
            pathserver = "DFW-SIP-DALEQ-TEST-TRPSBC1"
		    datacenter = "SAT"
		    servertype = "Transport"
	elseif path ~=nil and string.find(path,"SAT-SIP-CU1-GADSBC1") then
            pathName = "vis_sip_realm_method"
            event.log.ath_server = "DFW-SIP-DALEQ-TEST-TRPSBC1"
		    datacenter = "SAT"
		    servertype = "Transport"
	elseif path ~=nil and string.find(path,"SAT-SIP-CU1_TRPSBC1_SAT_SIP_CU1_TRPSBC2") then
            pathName = "vis_sip_realm_method"
            pathserver= "SAT-SIP-CU1-TRPSBC1_SAT-SIP-CU1-TRPSBC2"
		    datacenter = "SAT"
		    servertype = "Transport"
	elseif path ~=nil and string.find(path,"sat_itc3_test_sbc2")  then
            pathName = "vis_sip_realm_method"
            pathserver= "sat-itc3-test-sbc2"
		    datacenter = "SAT"
		    servertype = "Transport"
	elseif path ~=nil and string.find(path,"SAT_SIP_CU1_REGSBC1_SAT_SIP_CU1_REGSBC2")  then
            pathName = "vis_sip_realm_method"
            pathserver= "SAT-SIP-CU1-REGSBC1_SAT-SIP-CU1-REGSBC2"
		    datacenter = "SAT"
		    servertype = "Transport"
	elseif path ~=nil and string.find(path,"SAT_SIP_CU1_REGSBC1")  then
            pathName = "vis_sip_realm_method"
            pathserver= "SAT-SIP-CU1-REGSBC1"
		    datacenter = "SAT"
		    pevent.log.ath_server_type = "Transport"
	elseif path ~=nil and string.find(path,"SAT_SIP_CU1_TEST_SR1_SAT_SIP_CU1_TEST_SR2")  then
            pathName = "vis_sip_realm_method"
            pathserver= "SAT-SIP-CU1-TEST-SR1_SAT-SIP-CU1-TEST-SR2"
		    datacenter = "SAT"
		    servertype = "Transport"
       end
    end
end
-------------------------------------------------------------
-- This Function will recieve an object containing the event
-- log messages and will parse them into an array based on a
-- comma delimeter
-------------------------------------------------------------
function setValuesArray(csvFile)
    for line in csvFile:gmatch("[^\\\n]+") do
        valuesArray[rowNum] = {}
        local colNum = 1
        for cellVal in line:gmatch("[^,]+") do
            if(cellVal ~= nil) then
                valuesArray[rowNum][colNum] = cellVal
            else
			    valuesArray[rowNum][colNum] = 0
            end
            colNum = colNum + 1
        end
        rowNum = rowNum + 1
    end
end
function shutdown(emit)
    valuesArray = {}
    rowNum = 1
    rowIndx = -1
    colIndx = -1
end

function process(event, emit)               -- main processing
    csvFile = event.log.message
    path = event.log.file
    fields = csv.openstring(event.log.message):lines()()
    event.log.message = nil   
    
    if csvFile ~= nil and path ~=nil  then                  
        setFileParms(path)
        setValuesArray(csvFile)
        if path ~=nil then
            event.log.path_name = pathName
            event.log.path_server = pathserver
            event.log.path_datacenter = datacenter
            event.log.path_server_type = servertype
        end
        for rowIndx = 2,#valuesArray,1 do
            realm = ""
            for colIndx = 2,#valuesArray[rowIndx],1 do
                if valuesArray[rowIndx][colIndx] ~= nil and not string.find(valuesArray[rowIndx][colIndx],"-") and valuesArray[rowIndx][colIndx] ~= "0"  then
                    if tonumber(valuesArray[rowIndx][colIndx]) ~= nil  then -- is numeric
                        if #realm > 0 then
                            event.log.realm = realm
                        end
                        if valuesArray[1][colIndx] ~= nil then
                            event.log.header = valuesArray[1][colIndx]
                            event.log.value = valuesArray[rowIndx][colIndx]
                            emit(event)
                        end
                    else
                        if #realm > 0 then
                            realm = realm .. "_" .. valuesArray[rowIndx][colIndx] 
                        else
                            realm = valuesArray[rowIndx][colIndx]
                        end
                    end
                end
               
            end
        end
        
    end
    processedPath = event.log.file
end
"""
hooks.init = "init"
hooks.process = "process"
hooks.shutdown = "shutdown"

[transforms.vectormetric]
 type = "log_to_metric"
 inputs = ["vectorlua"]
 [[transforms.vectormetric.metrics]]
  type = "gauge"
  field = "value"
  name = "{{path_name}}"
  [transforms.vectormetric.metrics.tags]
    message_realm = "{{realm}}"
    message_header = "{{header}}"
    server = "{{path_server}}"
	datacenter = "{{path_datacenter}}"
	server_type = "{{path_server_type}}"
`

sink

[sinks.vectorConsole]
   type = "prometheus_exporter"
   inputs = ["vectormetric"]
   address = "0.0.0.01234"
   suppress_timestamp = true
   flush_period_secs = 86400 

[Manniac07]

@jszwedko what it keeps doing is loading the whole file for every line instead of spinning through the table i made and going on to next file

[jszwedko]

Thanks! That behavior looks like what I would expect based on this configuration. The process hook runs on each incoming line.

[pront]

The file source normally emits one event per line. Its multiline patterns match those lines, so \A and \z do not identify the beginning and end of the file.

There is also a separate issue in the Lua code: valuesArray and rowNum are initialized only in init, but every process call loops over all rows accumulated so far. That emits earlier rows again. I reproduced this with your accumulation loop on Vector 0.60.0. Processing each complete CSV event with a fresh table avoids that repetition; resetting the table for every individual line would lose the header instead.

As this is an older discussion, we’re closing it as part of the backlog cleanup. If you still need help, you’re very welcome to open a new discussion with your current configuration and a sample CSV. Please link back here so we can pick up from this context.