Мне нужно создать файл паркета из файлов csv, используя настроенный файл схемы json, например этот:
{"type" : "struct","fields" : [ {"name" : "tenor_bank","type" : "string","nullable" : false}, {"name":"tenor_frtb", "type":"string", "nullable":false}, {"name":"weight", "type":"decimal(25,5)", "nullable":false} ]}
Обратите внимание на поле веса.
Вот так выглядит входной CSV-файл:
tenor_1;tenor_2;weight
1D;3M;1
7D;3M;1
1W;3M;1
1OD;3M;1
14D;3M;1
2W;3M;1
21D;3M;1
3W;3M;1
28D;3M;1
30D;3M;1
1M;3M;1
56D;3M;1
60D;3M;1
2M;3M;1
61D;3M;1
84D;3M;1
90D;3M;1
3M;3M;1
91D;3M;1
92D;3M;1
112D;3M;0.8333
112D;6M;0.1667
Вот как я загружаю файл схемы json с его DataFrame:
val path: Path = new Path(mra_schema_parquet)
val fileSystem = path.getFileSystem(sc.hadoopConfiguration)
val inputStream: FSDataInputStream = fileSystem.open(path)
val schema_json = Stream.cons(inputStream.readLine(), Stream.continually( inputStream.readLine))
logger.debug("schema_json looks like " + schema_json.head)
val mySchemaStructType = DataType.fromJson(schema_json.head).asInstanceOf[StructType]
logger.debug("mySchemaStructType is " + mySchemaStructType)
myDF = loadCSV(sqlContext, path_input_csv,separator,mySchemaStructType ,header)
logger.debug("myDF.schema.json looks like " + myDF.schema.json)
inputStream.close()
//finally I create the parquet file. This line provokes the NuumberFormatException, concretely the line with .parquet(pathParquet)
writeDataFrame2Parquet(myDF, path_output_parquet, saveMode,header,separator)
//some utilities
def loadCSV(sqlContext : SQLContext, pathCSV: String, separator: String, customSchema: StructType, haveSchema: String): DataFrame = {
logger.info("loadCSV. header is " + haveSchema.toString + ", inferSchema is false pathCSV is " + pathCSV + " separator is " + separator)
sqlContext.read
.format("com.databricks.spark.csv")
.option("header", haveSchema) // Use first line of all files as header
.option("delimiter", separator)
.option("nullValue","")
//Esto provoca que pete en runtime si encuentra un fallo en la línea que esté parseando
.option("mode","FAILFAST")
.schema(customSchema)
.load(pathCSV)
}
def writeDataFrame2Parquet(df: DataFrame, pathParquet: String, saveMode: SaveMode,header: String,delimiter:String): Unit = {
df.write
.format("com.databricks.spark.csv")
.option("header", header)
.option("delimiter",delimiter)
.option("nullValue","")
.mode(saveMode)
.parquet(pathParquet)
}
Когда выполнение достигает последней строки, .parquet (pathParquet), происходит исключение:
Caused by: org.apache.spark.SparkException: Task failed while writing rows.
at org.apache.spark.sql.execution.datasources.DefaultWriterContainer.writeRows(WriterContainer.scala:250)
at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1$$anonfun$apply$mcV$sp$3.apply(InsertIntoHadoopFsRelation.scala:150)
at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1$$anonfun$apply$mcV$sp$3.apply(InsertIntoHadoopFsRelation.scala:150)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
at org.apache.spark.scheduler.Task.run(Task.scala:88)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
**Caused by: java.lang.NumberFormatException**
at java.math.BigDecimal.<init>(BigDecimal.java:545)
at java.math.BigDecimal.<init>(BigDecimal.java:739)
at com.databricks.spark.csv.util.TypeCast$.castTo(TypeCast.scala:68)
at com.databricks.spark.csv.CsvRelation$$anonfun$buildScan$2.apply(CsvRelation.scala:121)
at com.databricks.spark.csv.CsvRelation$$anonfun$buildScan$2.apply(CsvRelation.scala:108)
at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
at org.apache.spark.sql.execution.datasources.DefaultWriterContainer.writeRows(WriterContainer.scala:240)
... 8 more
Похоже, что когда spark-csv пытается отобразить поле "вес" как десятичное (25,5), библиотека вылетает. Кто-нибудь может мне помочь?
Спасибо.
