Исключение в потоке «главная» организация. apache .spark. sql .AnalysisException: Нет такого структурного поля startId в _1, _2;

Question

0

У меня есть 2 класса дел, как показано ниже:

  case class IcijAddressRaw(
                             node_id: Option[Long],
                             name: Option[String],
                             address: Option[String],
                             country_codes: Option[String],
                             countries: Option[String],
                             sourceID: Option[String],
                             valid_until: Option[String],
                             note: Option[String]
                           )

  case class IcijEdgesRaw(
                           START_ID: Option[Long],
                           TYPES: Option[String],
                           END_ID: Option[Long],
                           link: Option[String],
                           start_date: Option[java.sql.Date],
                           end_date: Option[java.sql.Date],
                           sourceID: Option[String],
                           valid_until: Option[String]
                         )

Я объединяю оба набора данных с классом дела, как указано ниже:

val addressWithEdgesDS = addressRawDS
  .joinWith(edgesRawDS, edgesRawDS("END_ID") === addressRawDS("node_id"), "inner")

val addressGroupDS = addressWithEdgesDS.groupByKey { fullAddress => fullAddress}.mapGroups {
  case (startId, fullAddress) =>
    (startId, fullAddress.toSeq)
}


def thirdPartyWithAddressDS = thirdPartyDS
  .joinWith(addressGroupDS, thirdPartyDS("thirdPartyId") === addressGroupDS("_2.startId"), "left_outer")
  .map{
    case (thirdParty,null) => thirdParty
    case (thirdParty, (thirdPartyId, addressSeq)) => thirdParty.copy(addresses = addressSeq.map(addressCaseClassMap[ThirdPartyCC]))
  }

I может построить банку без проблем. Однако, когда я запускаю следующий скрипт:

 ./runQSS.sh -s com.quantexa.academytask.etl.projects.icij.CreateCaseClass -e local -r etl.icij -c /home/training-admin/academy-task/application.conf

Ошибка:

Exception in thread "main" org.apache.spark.sql.AnalysisException: No such struct field startId in _1, _2;

Я пробовал имя столбца "START_ID", а также "_2.START_ID", но та же ошибка. Насколько я понимаю, эта структура объединенных наборов данных должна использовать _1 или _2 для ссылки на левый или правый столбец набора данных. Но в этом случае кажется, что не удалось указать правильный столбец.

моя присоединенная схема, как показано ниже: scala> addressWithEdgesDS.printSchema () root

 |-- _1: struct (nullable = false)
 |    |-- node_id: integer (nullable = true)
 |    |-- name: string (nullable = true)
 |    |-- address: string (nullable = true)
 |    |-- country_codes: string (nullable = true)
 |    |-- countries: string (nullable = true)
 |    |-- sourceID: string (nullable = true)
 |    |-- valid_until: string (nullable = true)
 |    |-- note: string (nullable = true)
 |-- _2: struct (nullable = false)
 |    |-- START_ID: integer (nullable = true)
 |    |-- TYPES: string (nullable = true)
 |    |-- END_ID: integer (nullable = true)
 |    |-- link: string (nullable = true)
 |    |-- start_date: string (nullable = true)
 |    |-- end_date: string (nullable = true)
 |    |-- sourceID: string (nullable = true)
 |    |-- valid_until: string (nullable = true)