如何在 Spark 中创建 Schema 文件答案

【问题标题】：How to create a Schema file in Spark如何在 Spark 中创建 Schema 文件
【发布时间】：2018-11-03 04:15:08
【问题描述】：

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

【解决方案1】：

要从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")

希望这会有所帮助！

【讨论】：

谢谢@ShankarKiorala 如果输入模式类似于balance decimal(10,0)。如何更新匹配函数中的代码？因为小数很常见，但是不同字段的进动会有所不同，比如case "decimal(10,0)" => DecimalType(10,0)
@BalakrishnanRamasamy 我认为像case "decimal(10,0)" => { val decimal = raw.split("(")(1).replace(")", "").split(",") DecimalType(decimal(0).toInt, decimal(1).toInt) } 这样的东西应该可以工作，但还没有经过测试。
谢谢@ShankarKoirala，我的主要问题是创建一个match 案例。假设，架构文件是 balance decimal(10,0) amount decimal(20,1) 在这种情况下，我的匹配案例看起来如何？
我不确定这个，但匹配数据类型小数的第一部分应该可以。

【解决方案2】：

您可以按以下格式创建名为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")

【讨论】：

【解决方案3】：

这样的东西更健壮一些，因为它使用了 hive 元存储：

    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
    }
``

【讨论】：

【解决方案4】：

您可以像这样指定架构：

import org.apache.spark.sql.types.{StructType, StructField, StringType,IntegerType};

例如：

val schema = new StructType(
Array(
   StructField("Age",IntegerType,true),
  StructField("Name",StringType,true),
  )
)

val data = spark.read.option("header", "false").schema(schema).csv("filename.csv")
data.show()

这将直接在数据框中创建它

【讨论】：