我正在尝试读取一个Schema文件(这是一个文本文件),并将它应用到我的CSV文件中,而没有头文件。因为我已经有了一个模式文件,所以我不想使用InferSchema
选项,这是一种开销。
我的输入模式文件如下,
"num IntegerType","letter StringType"
我正在尝试下面的代码来创建一个模式文件,
val schema_file = spark.read.textFile("D:\\Users\\Documents\\schemaFile.txt")
val struct_type = schema_file.flatMap(x => x.split(",")).map(b => (b.split(" ")(0).stripPrefix("\"").asInstanceOf[String],b.split(" ")(1).stripSuffix("\"").asInstanceOf[org.apache.spark.sql.types.DataType])).foreach(x=>println(x))
我得到的错误如下
Exception in thread "main" java.lang.UnsupportedOperationException: No Encoder found for org.apache.spark.sql.types.DataType
并尝试将其作为模式文件使用,同时使用如下所示的spark.read.csv
并将其写入ORC文件
val df=spark.read
.format("org.apache.spark.csv")
.option("header", false)
.option("inferSchema", true)
.option("samplingRatio",0.01)
.option("nullValue", "NULL")
.option("delimiter","|")
.schema(schema_file)
.csv("D:\\Users\\sampleFile.txt")
.toDF().write.format("orc").save("D:\\Users\\ORC")
需要帮助将文本文件转换为架构文件,并将输入的CSV文件转换为ORC。
发布于 2018-05-24 05:29:20
若要从text
文件创建架构,请创建一个match
type
函数,并将DataType
返回为
def getType(raw: String): DataType = {
raw match {
case "ByteType" => ByteType
case "ShortType" => ShortType
case "IntegerType" => IntegerType
case "LongType" => LongType
case "FloatType" => FloatType
case "DoubleType" => DoubleType
case "BooleanType" => BooleanType
case "TimestampType" => TimestampType
case _ => StringType
}
}
现在,通过将架构文件读取为
val schema = Source.fromFile("schema.txt").getLines().toList
.flatMap(_.split(",")).map(_.replaceAll("\"", "").split(" "))
.map(x => StructField(x(0), getType(x(1)), true))
现在,将csv文件读入
spark.read
.option("samplingRatio", "0.01")
.option("delimiter", "|")
.option("nullValue", "NULL")
.schema(StructType(schema))
.csv("data.csv")
希望这能有所帮助!
发布于 2020-10-08 10:02:37
您可以以以下格式创建名为schema.json
的JSON文件
{
"fields": [
{
"metadata": {},
"name": "first_fields",
"nullable": true,
"type": "string"
},
{
"metadata": {},
"name": "double_field",
"nullable": true,
"type": "double"
}
],
"type": "struct"
}
从读取此文件中创建结构架构
rdd = spark.sparkContext.wholeTextFiles("s3://<bucket>/schema.json")
text = rdd.collect()[0][1]
dict = json.loads(str(text))
custom_schema = StructType.fromJson(dict)
之后,可以使用struct作为模式来读取csv文件。
val df=spark.read
.format("org.apache.spark.csv")
.option("header", false)
.option("inferSchema", true)
.option("samplingRatio",0.01)
.option("nullValue", "NULL")
.option("delimiter","|")
.schema(custom_schema)
.csv("D:\\Users\\sampleFile.txt")
.toDF().write.format("orc").save("D:\\Users\\ORC")
发布于 2019-01-13 22:35:01
像这样的东西更健壮一些,因为它使用了蜂巢亚稳态:
import org.apache.hadoop.hive.metastore.api.FieldSchema
def sparkToHiveSchema(schema: StructType): List[FieldSchema] ={
schema.map(field => new FieldSchema(field.name,field.dataType.catalogString,field.getComment.getOrElse(""))).toList
}
``
https://stackoverflow.com/questions/50500804
复制相似问题