создать схему элемента массива в scala

Question

Я новичок в scala и пытаюсь создать собственную схему из массива элементов для чтения файлов на основе новой пользовательской схемы.

Я прочитал массивы из файла json и использовал метод разнесения и создал фрейм данных для каждого элемента в массиве столбцов.

val otherPeople = sqlContext.read.option("multiline", "true").json(otherPeopleDataset)
val column_values = otherPeople.withColumn("columns", explode($"columns")).select("columns.*")
column_values.printSchema()

Получен вывод:

column_values: org.apache.spark.sql.DataFrame = [column_id: string, data_sensitivty: string ... 3 more fields]
root
 |-- column_id: string (nullable = true)
 |-- data_sensitivty: string (nullable = true)
 |-- datatype: string (nullable = true)
 |-- length: string (nullable = true)
 |-- name: string (nullable = true)

val column_name = column_values.select("name","datatype")

column_name: org.apache.spark.sql.DataFrame = [name: string, datatype: string]

column_name.show(4)


+-----------------+--------+
|             name|datatype|
+-----------------+--------+
|    object_number| varchar|
|    function_type| varchar|
|            hof_1| varchar|
|            hof_2| varchar|
|           region| varchar|
|          country| varchar|
+-----------------+--------+

Теперь для всех значений, перечисленных выше, я хотел создать val Схема динамически.

пример:

val schema = new StructType()
      .add("object_number",StringType,true)
      .add("function_type",StringType,true)
      .add("hof_1",StringType,true)
      .add("hof_2",StringType,true)
      .add("region",StringType,true)
      .add("Country",StringType,true)

Я хочу динамически строить вышеупомянутую структуру, как только я получил столбец данных, я прочитал, что сначала мне нужно создать карту типа данных для каждого элемента, а затем создать структуру в l oop. Может ли кто-нибудь помочь здесь, поскольку у меня ограниченные знания scala.

Answer 1

DataFrame с полями позволяет собирать данные, и для каждого поля строки добавляется «StructType»:

val schemaColumns = column_name.collect()
val schema = schemaColumns.foldLeft(new StructType())(
  (schema, columnRow) => schema.add(columnRow.getAs[String]("name"), getFieldType(columnRow.getAs[String]("datatype")), true)
  )

def getFieldType(typeName: String): DataType = typeName match {
    case "varchar" => StringType
    // TODO include other types here
    case _ => StringType
  }

Answer 2

Вы можете следовать этому подходу, он может хорошо работать для вашего примера:

 //The schema is encoded in a string
  val schemaString = "object_number function_type hof_1 hof_2 region Country"
  //Generate the schema based on the string of schema
  val fields = schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, nullable = true))
  val schema = StructType(fields)
  //Convert records of the RDD (myRdd) to Rows
  val rowRDD = sc.textFile("dir").map(line => line.split(",")).map(attributes => Row(attributes(0),attributes(1),attributes(2), attributes(3),attributes(4),attributes(5)))
  //Apply the schema to the RDD
  val perDF = spark.createDataFrame(rowRDD, schema)

Надеюсь, это даст вам некоторые подсказки, с уважением.