I am trying to use the AWS Glue ETL tool to parse a CSV file to Parquet. I am following this tutorial from the Amazon Website.
Since I have a bunch of CSV files on a S3 bucket, I am keeping their content compressed as GZIP. I am using the following lambda function to download the data from a zip, extract the CSV files and save it as a compressed CSV on my S3 bucket:
import json
import boto3
from botocore.vendored import requests
import zipfile
from gzip import GzipFile
from io import BytesIO
def lambda_handler(event, context):
s3 = boto3.resource('s3')
bucket_name = "my-bucket"
file_url = 'http://dados.cvm.gov.br/dados/FI/DOC/CDA/DADOS/'
file_name = "cda_fi_201801"
req = requests.get(file_url + file_name+".zip", stream=True)
data = req.raw.read()
zf = zipfile.ZipFile(BytesIO(data))
for fn in zf.namelist():
bytes = zf.read(fn).decode("windows-1252")
print ('File:', fn)
print ('has',len(bytes),'bytes')
# Choose folder name to put csv file
parts = fn.split("_")
folder = "PL"
if(parts[2] == "BLC"):
folder = "BLC_"+parts[3]
# BytesIO to not save to disk
gz_body = BytesIO()
gz = GzipFile(None, 'wb', 9, gz_body)
# Write csv bytes to gzip body
gz.write(bytes.encode('utf8'))
s3.Bucket(bucket_name).put_object(Key=folder + "/" + fn,
ContentType="text/plain",
ContentEncoding='gzip',
Body=gz_body.getvalue())
gz.close()
return {
'statusCode': 200,
}
Running the AWS Glue Crawler on blc_1 folder, I got the following table properties:
{
"StorageDescriptor": {
"cols": {
"FieldSchema": [
{
"name": "tp_fundo",
"type": "string",
"comment": ""
},
{
"name": "cnpj_fundo",
"type": "string",
"comment": ""
},
{
"name": "denom_social",
"type": "string",
"comment": ""
},
{
"name": "dt_comptc",
"type": "string",
"comment": ""
},
{
"name": "tp_aplic",
"type": "string",
"comment": ""
},
{
"name": "tp_ativo",
"type": "string",
"comment": ""
},
{
"name": "emissor_ligado",
"type": "string",
"comment": ""
},
{
"name": "tp_negoc",
"type": "string",
"comment": ""
},
{
"name": "qt_venda_negoc",
"type": "double",
"comment": ""
},
{
"name": "vl_venda_negoc",
"type": "double",
"comment": ""
},
{
"name": "qt_aquis_negoc",
"type": "double",
"comment": ""
},
{
"name": "vl_aquis_negoc",
"type": "double",
"comment": ""
},
{
"name": "qt_pos_final",
"type": "double",
"comment": ""
},
{
"name": "vl_merc_pos_final",
"type": "double",
"comment": ""
},
{
"name": "vl_custo_pos_final",
"type": "string",
"comment": ""
},
{
"name": "dt_confid_aplic",
"type": "string",
"comment": ""
},
{
"name": "tp_titpub",
"type": "string",
"comment": ""
},
{
"name": "cd_isin",
"type": "string",
"comment": ""
},
{
"name": "cd_selic",
"type": "bigint",
"comment": ""
},
{
"name": "dt_emissao",
"type": "string",
"comment": ""
},
{
"name": "dt_venc",
"type": "string",
"comment": ""
}
]
},
"location": "s3://my-bucket/BLC_1/",
"inputFormat": "org.apache.hadoop.mapred.TextInputFormat",
"outputFormat": "org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat",
"compressed": "true",
"numBuckets": "-1",
"SerDeInfo": {
"name": "",
"serializationLib": "org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe",
"parameters": {
"field.delim": ";"
}
},
"bucketCols": [],
"sortCols": [],
"parameters": {
"skip.header.line.count": "1",
"sizeKey": "731056",
"objectCount": "1",
"UPDATED_BY_CRAWLER": "blc-1",
"CrawlerSchemaSerializerVersion": "1.0",
"recordCount": "1884",
"averageRecordSize": "258",
"CrawlerSchemaDeserializerVersion": "1.0",
"compressionType": "gzip",
"classification": "csv",
"columnsOrdered": "true",
"areColumnsQuoted": "false",
"delimiter": ";",
"typeOfData": "file"
},
"SkewedInfo": {},
"storedAsSubDirectories": "false"
},
"parameters": {
"skip.header.line.count": "1",
"sizeKey": "731056",
"objectCount": "1",
"UPDATED_BY_CRAWLER": "blc-1",
"CrawlerSchemaSerializerVersion": "1.0",
"recordCount": "1884",
"averageRecordSize": "258",
"CrawlerSchemaDeserializerVersion": "1.0",
"compressionType": "gzip",
"classification": "csv",
"columnsOrdered": "true",
"areColumnsQuoted": "false",
"delimiter": ";",
"typeOfData": "file"
}
}
After that, I followed the tutorial and tried to run an ETL job with the following script automatically generated by Glue:
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
## @type: DataSource
## @args: [database = "cvm", table_name = "blc_1", transformation_ctx = "datasource0"]
## @return: datasource0
## @inputs: []
datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "cvm", table_name = "blc_1", transformation_ctx = "datasource0")
## @type: ApplyMapping
## @args: [mapping = [("tp_fundo", "string", "tp_fundo", "string"), ("cnpj_fundo", "string", "cnpj_fundo", "string"), ("denom_social", "string", "denom_social", "string"), ("dt_comptc", "string", "dt_comptc", "string"), ("tp_aplic", "string", "tp_aplic", "string"), ("tp_ativo", "string", "tp_ativo", "string"), ("emissor_ligado", "string", "emissor_ligado", "string"), ("tp_negoc", "string", "tp_negoc", "string"), ("qt_venda_negoc", "double", "qt_venda_negoc", "double"), ("vl_venda_negoc", "double", "vl_venda_negoc", "double"), ("qt_aquis_negoc", "double", "qt_aquis_negoc", "double"), ("vl_aquis_negoc", "double", "vl_aquis_negoc", "double"), ("qt_pos_final", "double", "qt_pos_final", "double"), ("vl_merc_pos_final", "double", "vl_merc_pos_final", "double"), ("vl_custo_pos_final", "string", "vl_custo_pos_final", "string"), ("dt_confid_aplic", "string", "dt_confid_aplic", "string"), ("tp_titpub", "string", "tp_titpub", "string"), ("cd_isin", "string", "cd_isin", "string"), ("cd_selic", "long", "cd_selic", "long"), ("dt_emissao", "string", "dt_emissao", "string"), ("dt_venc", "string", "dt_venc", "string")], transformation_ctx = "applymapping1"]
## @return: applymapping1
## @inputs: [frame = datasource0]
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("tp_fundo", "string", "tp_fundo", "string"), ("cnpj_fundo", "string", "cnpj_fundo", "string"), ("denom_social", "string", "denom_social", "string"), ("dt_comptc", "string", "dt_comptc", "string"), ("tp_aplic", "string", "tp_aplic", "string"), ("tp_ativo", "string", "tp_ativo", "string"), ("emissor_ligado", "string", "emissor_ligado", "string"), ("tp_negoc", "string", "tp_negoc", "string"), ("qt_venda_negoc", "double", "qt_venda_negoc", "double"), ("vl_venda_negoc", "double", "vl_venda_negoc", "double"), ("qt_aquis_negoc", "double", "qt_aquis_negoc", "double"), ("vl_aquis_negoc", "double", "vl_aquis_negoc", "double"), ("qt_pos_final", "double", "qt_pos_final", "double"), ("vl_merc_pos_final", "double", "vl_merc_pos_final", "double"), ("vl_custo_pos_final", "string", "vl_custo_pos_final", "string"), ("dt_confid_aplic", "string", "dt_confid_aplic", "string"), ("tp_titpub", "string", "tp_titpub", "string"), ("cd_isin", "string", "cd_isin", "string"), ("cd_selic", "long", "cd_selic", "long"), ("dt_emissao", "string", "dt_emissao", "string"), ("dt_venc", "string", "dt_venc", "string")], transformation_ctx = "applymapping1")
## @type: ResolveChoice
## @args: [choice = "make_struct", transformation_ctx = "resolvechoice2"]
## @return: resolvechoice2
## @inputs: [frame = applymapping1]
resolvechoice2 = ResolveChoice.apply(frame = applymapping1, choice = "make_struct", transformation_ctx = "resolvechoice2")
## @type: DropNullFields
## @args: [transformation_ctx = "dropnullfields3"]
## @return: dropnullfields3
## @inputs: [frame = resolvechoice2]
dropnullfields3 = DropNullFields.apply(frame = resolvechoice2, transformation_ctx = "dropnullfields3")
## @type: DataSink
## @args: [connection_type = "s3", connection_options = {"path": "s3://my-bucket/blc_1"}, format = "parquet", transformation_ctx = "datasink4"]
## @return: datasink4
## @inputs: [frame = dropnullfields3]
datasink4 = glueContext.write_dynamic_frame.from_options(frame = dropnullfields3, connection_type = "s3", connection_options = {"path": "s3://my-bucket/blc_1"}, format = "parquet", transformation_ctx = "datasink4")
job.commit()
I got the following error:
19/03/27 19:10:07 WARN TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0, ip-172-32-89-229.us-east-2.compute.internal, executor 1): com.amazonaws.services.glue.util.FatalException: Unable to parse file: cda_fi_BLC_1_201801.csv
at com.amazonaws.services.glue.readers.JacksonReader.hasNextFailSafe(JacksonReader.scala:94) at com.amazonaws.services.glue.readers.JacksonReader.hasNext(JacksonReader.scala:38) at com.amazonaws.services.glue.hadoop.TapeHadoopRecordReader.nextKeyValue(TapeHadoopRecordReader.scala:63) at org.apache.spark.rdd.NewHadoopRDD$$anon$1.hasNext(NewHadoopRDD.scala:207) at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37) at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:148) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96) at org.apache.spark.scheduler.ShuffleMapTask.
Unfortunately, it gives no clue of what may be happening. I was able to perform the ETL when the CSV content was not compressed, so my best guess is that I am doing something wrong on the gzip compression or there is some configuration missing.
If you have an idea of what is happening, I appreciate some help.
Thanks!
