社区首页 >问答首页 >如何在星火中创建架构文件

问如何在星火中创建架构文件
EN

Stack Overflow用户

提问于 2018-05-23 20:08:55

回答 4查看 6.7K关注 0票数 5

我正在尝试读取一个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

字段(类："org.apache.spark.sql.types.DataType"，名称："_2") -根类："scala.Tuple2“

并尝试将其作为模式文件使用，同时使用如下所示的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。

scala

apache-spark-sql

schema

orc

回答 4

Stack Overflow用户

回答已采纳

发布于 2018-05-23 21: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")

希望这能有所帮助！

票数 6

Stack Overflow用户

发布于 2020-10-08 02: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")

票数 6

Stack Overflow用户

发布于 2019-01-13 14: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
    }
``