Ошибка при добавлении нового строкового столбца utf8 в строку в искре Scala

Question

Я пытаюсь добавить новый столбец в каждую строку DataFrame, как это

  def addNamespace(iter: Iterator[Row]): Iterator[Row] = {
    iter.map (row => {
      println(row.getString(0))
//      Row.fromSeq(row.toSeq ++ Array[String]("shared"))

      val newseq = row.toSeq ++ Array[String]("shared")
      Row(newseq: _*)
    })
    iter
  }

  def transformDf(source: DataFrame)(implicit spark: SparkSession): DataFrame = {
    val newSchema = StructType(source.schema.fields ++ Array(StructField("namespace", StringType, nullable = true)))
    val df = spark.sqlContext.createDataFrame(source.rdd.mapPartitions(addNamespace), newSchema)
    df.show()
    df

  }

Но я продолжаю получать эту ошибку - Caused by: java.lang.RuntimeException: org.apache.spark.unsafe.types.UTF8String is not a valid external type for schema of string в строке df.show()

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

Я также пытался val again = sourceDF.withColumn("namespace", functions.lit("shared")), но у него та же проблема.

Схема уже прочитанных данных

root
 |-- name: string (nullable = true)
 |-- data: struct (nullable = true)
 |    |-- name: string (nullable = true)
 |    |-- description: string (nullable = true)
 |    |-- activates_on: timestamp (nullable = true)
 |    |-- expires_on: timestamp (nullable = true)
 |    |-- created_by: string (nullable = true)
 |    |-- created_on: timestamp (nullable = true)
 |    |-- updated_by: string (nullable = true)
 |    |-- updated_on: timestamp (nullable = true)
 |    |-- properties: map (nullable = true)
 |    |    |-- key: string
 |    |    |-- value: string (valueContainsNull = true)

Answer 1

Причина: java.lang.RuntimeException: org.apache.spark.unsafe.types.UTF8String не является допустимым внешним типом для схемы строки

означает, что его невозможно понять как строковый тип ... для недавно добавленного столбца «пространства имен».

Четко указывает на ошибку несоответствия типов данных на уровне катализатора ...

см. Искровой код здесь ..

override def eval(input: InternalRow): Any = {
    val result = child.eval(input)
    if (checkType(result)) {
      result
    } else {
      throw new RuntimeException(s"${result.getClass.getName}$errMsg")
    }
  }

и сообщение об ошибке s" is not a valid external type for schema of ${expected.catalogString}"

Таким образом, строка UTF не является реальной строкой, поэтому вам необходимо кодировать / декодировать ее перед передачей в качестве типа строки, иначе катализатор не сможет понять, что вы передаете.

Как это исправить?

Ниже приводится содержание SO, в котором будет описано, как кодировать / декодировать в / из utfstring в строку и наоборот ... вам может потребоваться применить подходящее решение для этого.

https://stackoverflow.com/a/5943395/647053 строка декодирования utf-8

Примечание: Этот онлайновый инструмент кодирования / декодирования UTF-8 очень удобен для размещения образцов данных и их преобразования в строку. попробуйте сначала ....