LAE

Note: This was originally posted by an inactive account. Content was preserved by moving under an admin account.

Originally posted by: jpstory

Hi

The following node is created within Python Node to ingest csv files with column headers not on the first row.

The node skips first N rows and take the (N + 1)th row as column header and then parse the data.

It is working but considerably slower relative to lavaStorm CSV Node. (We are looking at 38 seconds vs. 5 seconds process time per file)

I need to ingest hundrunds of files worth 50+GB daily, it's super inefficient to manually open up each file and delete the report header, so that the data header can be on the first row.

However, we need help with improving our own customized CSV Node, please kindly advise on why it's so much slower than yours.



node:CSV_Node_3
bretype:core::Python
editor:Label=CSV Node
editor:sortkey=576b037f01e457bb_13
editor:icon=spreadsheet.ico
input:5627f61071af6efa/in1=
output:5627f61403a213c1/Part_1=
output:580684ca4c417a85/out2=
prop:DeleteSourceFile=False
prop:OutputToPath=C:\Test
prop:OverWrite=False
prop:Python2Implementation=<<EOX
import braininfo
import os
import csv
import codecs

def setup(brainNodeControlObj, BrainNodeClass):
class BrainNode(BrainNodeClass):
def initialize(self):

super(BrainNode, self).initialize()

#Load Parameters
try:
self.x = self.properties.getString("ls.brain.node.csvLineRe mover.RowsOfHeader")
except:
self.x = -1 # meaning let built-in logic determine how many rows to skip
self.OutputPath = self.properties.getString("ls.brain.node.csvLineRe mover.OutputToPath")
self.overWrite = self.properties.getBool("ls.brain.node.csvLineRemo ver.OverWrite")
self.deleteSource = self.properties.getBool("ls.brain.node.csvLineRemo ver.DeleteSourceFile")

#Read First File Name
self.inRec = self.inputs[0].read()
if not self.inRec:
return False
FileName = self.inRec["FileName"]

#Open First File in the Input Pin
with open(FileName,'rU') as Src:
csvReader = csv.reader(Src)
#Skip File Header Lines
for i in range(int(self.x)):
next(csvReader,None)
#Read Data Headers
self.Headers = []
HeaderLine = csvReader.next()
#Create Header List
h = 0
for Fld in HeaderLine:
h = h + 1
self.Headers.insert(h,str(Fld))

#Initialize Output Pin
metadata = self.newMetadata()
#Parse Headers into Output Pin
for h in HeaderLine:
metadata.append(str(h), "String")
metadata.append("FileName","String")
metadata.append("Check", "String")
self.outputs[0].metadata = metadata

#Initialize Exception Output Pin
metadataException = self.newMetadata()
metadataException.append("FileName", "String")
metadataException.append("Check", "String")
self.outputs[1].metadata = metadataException


def finalize(self, val):
super(BrainNode, self).finalize(val)

def pump(self, quant):
while quant.permitsRunning(self):

FileName = self.inRec["FileName"]

try:
success = self.processParse(FileName)
except:
outRec = self.outputs[1].newRecord()
outRec["FileName"] = FileName
self.outputs[1].write(outRec)

self.inRec = self.inputs[0].read()
if not self.inRec:
return False

#Remove Source File If Requested:
if self.deleteSource and success:
os.remove(FileName)
return True

def processParse(self, FileName):
success = False

#Open Input File
Src = open(FileName,'rU')
#Read The Data in The File
csvReader = csv.reader(Src)
#Skip File Header Line
for i in range(int(self.x)):
next(csvReader,None)
#Skip Data Header
SkipHeaderLine = csvReader.next()
#Loop Through Each Row of Data in the File
for row in csvReader:
outRec = self.outputs[0].newRecord()
h = -1
#Loop Through Each Field in The Row
for r in row:
h = h + 1
outRec[str(self.Headers[h])] = str(r)
self.outputs[0].write(outRec)
return True
return BrainNode
EOX
prop:RowsOfHeader=2
editor:XY=950,640
editor:propdef=RowsOfHeader|string|Parameter|ls.br ain.node.csvLineRemover.RowsOfHeader|None
editor:propdef=OutputToPath|string|Parameter|ls.br ain.node.csvLineRemover.OutputToPath|None
editor:propdef=OverWrite|string|Parameter|ls.brain .node.csvLineRemover.OverWrite|None
editor:propdef=DeleteSourceFile|string|Parameter|l s.brain.node.csvLineRemover.DeleteSourceFile|None
end:CSV_Node_3

Note: This was originally posted by an inactive account. Content was preserved by moving under an admin account.

Originally posted by: stonysmith

You should focus your efforts on this line of code:

outRec[str(self.Headers[h])] = str(r)

Looking up fields by column NAME is very slow. If you can convert this to using pre-computed numeric indexes, you will see a huge improvement in speed.

Note: This was originally posted by an inactive account. Content was preserved by moving under an admin account.

Originally posted by: jpstory

Thanks for the advice.

As you suggested, instead of using the actual field to search the target column, I used the index: outRec[h] = str(r).

Thanks to you I was able to boost the perform by cutting the ingestion time from 39 seconds down to 30 seconds.

Is there anything I can try to furhter boost the speed? Thank you in advance!

Note: This was originally posted by an inactive account. Content was preserved by moving under an admin account.

Originally posted by: stonysmith

See what happens if you move this up above the loop:
outRec = self.outputs[0].newRecord()

You don't have to create a new record every time.. it works fine if you just overlay the fiels in a single record.

That may not help much, but it's worth a try. Other than that, nothing else jumps out at me.. your next option is to try to add some timing so that you can find where the delay is happening.

Note: This was originally posted by an inactive account. Content was preserved by moving under an admin account.

Originally posted by: jpstory

Thanks.

I changed outRec[h] = str(r) to outRec[h] = r that cut the time down by another 2 seconds.

Just courious, how come lavastorm's CSV node can read the same file in just 5 seconds? What's the difference between the two nodes?