[
https://issues.apache.org/jira/browse/ARROW-16822?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17555276#comment-17555276
]
Mahesha Subrahamanya commented on ARROW-16822:
----------------------------------------------
we did try with traceback, exception handler however nothing worked here.
when pandas hand over the chunk data to pyarrow which is responsible to
converting into parquet file. we suspect during the process of converting it's
failing however it's not throwing the right error code/error message hence need
your help. kindly let me know if anything can be helpful is really appreciated.
since we are running into this issue we couldn't deliver this project as it's
dependency at the python libraries like pandas/pyarrow.
> Python Error: <>, exitCode: <139> when csv file converting parquet using
> pandas/pyarrow libraries
> -------------------------------------------------------------------------------------------------
>
> Key: ARROW-16822
> URL: https://issues.apache.org/jira/browse/ARROW-16822
> Project: Apache Arrow
> Issue Type: Bug
> Components: Parquet, Python
> Affects Versions: 5.0.0
> Reporter: Mahesha Subrahamanya
> Priority: Blocker
> Attachments: convertCSV2Parquet.png
>
>
> Our main requirement is to read source file (structured/semi structured
> /unstructured) which are residing in AWS s3 through AWS redshift database,
> where our customer have direct access to analyze the data very
> quickly/seamlessly for reporting purpose without defining the schema info for
> the file.
> We have created an data lake (aws s3) workspace where our customers dumps
> csv/parquet huge size files (like 10/15 GB). We have developed a framework
> which is consuming pandas/pyarrow (parquet) libraries to read source files in
> chunking manner and identifying schema meaning (datatype/length) and push it
> to AWS Glue where AWS redshift database can talk seamlessly to s3 files can
> read very quickly.
>
> Following is the snippet of parquet conversion where i'm getting this error.
> Please take a look
>
> read_csv_args = \{'filepath_or_buffer': src_object, 'chunksize':
> self.chunkSizeLimit, 'encoding': 'UTF-8','on_bad_lines': 'error','sep':
> fileDelimiter, 'low_memory': False, 'skip_blank_lines': True, 'memory_map':
> True} # 'verbose': True , In order to enable memory consumption logging
>
> if srcPath.endswith('.gz'):
> read_csv_args['compression'] = 'gzip'
> if fileTextQualifier:
> read_csv_args['quotechar'] = fileTextQualifier
> with pd.read_csv(**read_csv_args) as reader:
> for chunk_number, chunk in enumerate(reader, 1):
> # To support shape-shifting for the incoming datafiles,
> need to make sure match file with number of columns if not delete
> if glueMasterSchema is not None:
> sessionSchema=copy.deepcopy(glueMasterSchema)
> #copying using deepcopy() method
> chunk.columns = chunk.columns.str.lower() # modifying
> the column header of all columns to lowercase
> fileSchema = list(chunk.columns)
> for key in list(sessionSchema):
> if key not in fileSchema:
> del sessionSchema[key]
> fields = []
> for col,dtypes in sessionSchema.items():
> fields.append(pa.field(col, dtypes))
> glue_schema = pa.schema(fields)
> # To identify the boolean datatype and convert back
> to STRING which was done during the BF schema
> for cols in chunk.columns:
> try:
> if chunk[cols].dtype =='bool':
> chunk[cols] = chunk[cols].astype('str')
> if chunk[cols].dtype =='object':
> chunk[cols] =
> chunk[cols].fillna('').astype('str').tolist()
> except (ParserError,ValueError,TypeError):
> pass
> log.debug("chunk count", chunk_number, "chunk length",
> len(chunk), 'glue_schema', glue_schema, 'Wrote file', targetKey)
> #log.debug("during pandas chunk data ", chunk,"df
> schemas:", chunk.dtypes)
> table = pa.Table.from_pandas(chunk, schema=glue_schema ,
> preserve_index=False)
> log.info('Glue schema:',glue_schema,'for a
> [file:',targetKey|file:///',targetKey])
> log.info('pandas memory utilization during chunk process:
> ', chunk.memory_usage().sum(), 'Bytes.','\n\n\n')
> # Guess the schema of the CSV file from the first chunk
> #if pq_writer is None:
> if chunk_number == 1:
> #parquet_schema = table.schema
> # Open a Parquet file for writing
> pq_writer = pq.ParquetWriter(targetKey,
> schema=glue_schema, compression='snappy') # In PyArrow we use, Snappy
> generally results in better performance
> log.debug("table schema :",
> pprint.pformat(table.schema).replace('\n', ',').replace('\r', ','),' for:',
> inputFileName)
> # writing the log information into s3://etl_activity
> etlActivityLog.append({'tableObjectName':
> targetDirectory[:-1], 'sourceFileName': inputFileName, 'targetFileName':
> parquetFileName, 'message': 'File Converted Successfully', 'number of rows
> processed': str(table.num_rows), 'fileStatus': 'SUCCESS'})
> logInfo = self.read_logInfo(etlActivityLog)
> self.s3Handle.putObject(s3Client,
> 'etl_process_all.json', logInfo, bucketName, self.etlJobActivityLogFolder )
> # Write CSV chunk to the parquet file
> pq_writer.write_table(table)
> i += 1
> log.info( 'chunk count:', i, 'for a given
> [file:',targetKey,'whitelist:',targetDirectory[:-1|file:///',targetKey,'whitelist:',targetDirectory[:-1]])
> # Close a Parquet file writer
> if pq_writer is not None and pq_writer.is_open:
> pq_writer.close()
> pq_writer = None
> s3key = outputDirectory + targetDirectory + parquetFileName
> self.s3Handle.waitForFile(s3Client, bucketName, s3key)
> log.info('Metadata info:', table.column_names, 'number of
> columns:', table.num_columns, 'number of rows:', table.num_rows, 'Glue Object
> Name:', targetDirectory[:-1])
> log.debug('Wrote file', targetKey, 'with chunk count:',
> chunk_number)
> log.debug('Stream copy', targetKey, 'to parquet took:',
> datetime.now() - start_time)
> log.info('Final parquert convert:',sys.exc_info())
> except (EOFError, IOError) as x:
> log.error("error in source file for EOFError, IOError" % x)
> raise SystemExit('convert2Parquet EOFError:'+sys.exc_info())
> except (ValueError, ParserError) as x:
> log.error("error in source for ValueError, ParserError" % x)
> raise SystemExit('convert2Parquet valueError:'+sys.exc_info())
>
> finally:
> if pq_writer is not None and pq_writer.is_open:
> pq_writer.close()
>
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
