Как предоставить схему в spark / scala извне кода

Question

Я хочу прочитать файл для ex: schema_file , который будет содержать схему и использовать ее в коде для создания DataFrame

Я прочитал о ConfigFactory дляпредоставить схему, но не могу использовать ее, так как схема может быть изменена в будущем.

schema[
  {
     columnName = EXAMPLE_1
     type = string
  },
  {
     columnName = EXAMPLE_2
     type = string
  },
  {
     columnName = EXAMPLE_3
     type = string
  }
]

, если я использую это, чем мне нужно читать каждый columnName как

    config.getString("schema.ColumnName1")

, но столбцыне являются фиксированными, и количество столбцов может быть изменено.

Также я попытался использовать case class , но в этом также мне нужно указать каждое поле.

Может кто-нибудь сказать мне, как я могу прочитать схему из-за пределов кода.

Answer 1

Вы можете попробовать эту библиотеку, чтобы загрузить конфигурацию и отобразить ее в классы scala: https://github.com/pureconfig/pureconfig

Надеюсь, что-то подобное сработает для вас:

import scala.io.Source
import scala.util.parsing.combinator.syntactical.StandardTokenParsers

object Application extends App {
  override def main(args: Array[String]): Unit = {
    val fileContents = Source.fromFile("src/main/resources/schema_file").getLines.mkString
    print(ConfigDSL.parseSchema(fileContents))
  }
}

case class Schema(columns: List[Column])
case class Column(columnName: String, columnType: String)

object ConfigDSL extends StandardTokenParsers {
  lexical.delimiters ++= List("[", "]", "{", "}", ",", " ", "=", "\n")
  lexical.reserved ++= List("schema", "type", "columnName")

  def parseSchema(schemaString: String): Schema = 
    schema(new lexical.Scanner(schemaString)) match {
      case Success(columns, _) => Schema(columns)
      case Failure(msg, _) => throw new RuntimeException(msg)
      case Error(msg, _) => throw new RuntimeException(msg)
    }

  def schema: Parser[List[Column]] =
    "schema" ~ "[" ~ listOfColumns ~ "]" ^^ { case _ ~ _ ~ recipeList ~ _ => recipeList }

  def columnDefinition: Parser[Column] =
    "{" ~ "columnName" ~ "=" ~ ident ~ "type" ~ "=" ~ ident ~ "}" ^^ {
      case _ ~ _ ~ _ ~ column ~ _ ~ _ ~ columnType ~ _ => Column(column, columnType)
    }

  def listOfColumns: Parser[List[Column]] =
    repsep(columnDefinition, ",")  ^^ { stepList: List[Column] => stepList}
}