Многоуровневый комплексный Json с использованием Spark SQL

Question

У меня есть особое требование для преобразования некоторых связанных таблиц данных во вложенный json, как показано ниже, с использованием Spark SQL. Я добился этого с помощью Scala, но не смог решить его в Spark SQL.

{
"REPORTING_CARRIER": "9E", "DISTANCE": "3132", "ORIGIN_STATE_NM": "Пенсильвания", "КВАРТАЛ": "2", "ГОД": "2017", "ITIN_GEO_TYPE": "2", "BULK_FARE": "0", "ORIGIN": "АВА", "ORIGIN_AIRPORT_ID": "10135", "ITIN_FARE": "787", "ORIGIN_CITY_MARKET_ID": "30135", "Roundtrip": "1", "Рынок": [
{
"MKT_DISTANCE": "1566", "MKT_BULK_FARE": "0", "MKT_NO_OF_CPNS": "2", "MKT_DEST_STATE_NM": "Техас", "MKT_OP_CARR_GRP": "9E: DL", "MKT_TK_CARR_GRP": "DL: DL", "MKT_MILES_FLOWN": "1566", "MKT_AIRPORT_GROUP": "ABE: АТЛ: СБ", "MKT_FARE_AMT": "393,5", "MKT_ORIG_STATE_NM": "Пенсильвания", "MKT_DEST_ARPT_CITY_NM": "33214", "MKT_RPTG_CARR_NM": "9E", "MKT_DEST": "СБ", "MKT_DEST_CNTRY": "США", "MKT_ORIG_CNTRY": "США", "Купон": [
{
"CPN_STATE_NM": "Грузия", "CPN_DEST": "АТЛ", "CPN_TKT_CARR_NM": "DL", "TRIP_BREAK": "", "CPN_MKT_ORIG_ARPT_NM": "10135", "CLASS_OF_SVC": "X", "CPN_TKT_NBR": "2017245", "CPN_DEST_CITY_MKT_NM": "30397", "CPN_DISTANCE": "692", "Seq_num": "1", "ITIN_GEO_TYPE": "2", "CPN_RPTG_CARR_NM": "9E", "COUPON_GEO_TYPE": "2", "CPN_ORIG_STATE_NM": "Пенсильвания", "CPN_OPERG_CARR_NM": "9E", "CPN_ORIG": "АВА", "CPN_PASSENGERS": "1", "COUPON_TYPE": "А", "CPN_DEST_ARPT_NM": "10397", "CPN_MKT_ORIG_CITY_NM": "30135", "CPN_DEST_CNTRY": "США", "CPN_MKT_ID": "201724501", "CPN_ORIG_CNTRY": "США" }, {
"CPN_STATE_NM": "Техас", "CPN_DEST": "СБ", "CPN_TKT_CARR_NM": "DL", "TRIP_BREAK": "X", "CPN_MKT_ORIG_ARPT_NM": "10397", "CLASS_OF_SVC": "X", "CPN_TKT_NBR": "2017245", "CPN_DEST_CITY_MKT_NM": "33214", "CPN_DISTANCE": "874", "Seq_num": "2", "ITIN_GEO_TYPE": "2", "CPN_RPTG_CARR_NM": "9E", "COUPON_GEO_TYPE": "2", "CPN_ORIG_STATE_NM": "Грузия", "CPN_OPERG_CARR_NM": "DL", "CPN_ORIG": "АТЛ", "CPN_PASSENGERS": "1", "COUPON_TYPE": "А", "CPN_DEST_ARPT_NM": "14683", "CPN_MKT_ORIG_CITY_NM": "30397", "CPN_DEST_CNTRY": "США", "CPN_MKT_ID": "201724501", "CPN_ORIG_CNTRY": "США" } ], "MKT_ITIN_ID": "2017245", "MKT_OPERG_CARR_NM": "99", "MKT_DEST_ARPT_NM": "14683", "MKT_ORIG_ARPT_NM": "АВА", "MKT_ITIN_GEO_TYPE": "2", "MKT_PASSENGERS": "1", "MKT_ID": "201724501", "MKT_TKT_CARR_NM": "DL" }, {
"MKT_DISTANCE": "1566", "MKT_BULK_FARE": "0", "MKT_NO_OF_CPNS": "2", "MKT_DEST_STATE_NM": "Пенсильвания", "MKT_OP_CARR_GRP": "DL: DL", "MKT_TK_CARR_GRP": "DL: DL", "MKT_MILES_FLOWN": "1566", "MKT_AIRPORT_GROUP": "СБ: ATL: ABE", "MKT_FARE_AMT": "393,5", "MKT_ORIG_STATE_NM": "Техас", "MKT_DEST_ARPT_CITY_NM": "30135", "MKT_RPTG_CARR_NM": "9E", "MKT_DEST": "АВА", "MKT_DEST_CNTRY": "США", "MKT_ORIG_CNTRY": "США", "Купон": [
{
"CPN_STATE_NM": "Грузия", "CPN_DEST": "АТЛ", "CPN_TKT_CARR_NM": "DL","TRIP_BREAK": "", "CPN_MKT_ORIG_ARPT_NM": "14683", "CLASS_OF_SVC": "X", "CPN_TKT_NBR": "2017245", "CPN_DEST_CITY_MKT_NM": "30397", "CPN_DISTANCE": "874", "Seq_num": "3", "ITIN_GEO_TYPE": "2", "CPN_RPTG_CARR_NM": "9E", "COUPON_GEO_TYPE": "2", "CPN_ORIG_STATE_NM": "Техас", "CPN_OPERG_CARR_NM": "DL", "CPN_ORIG": "СБ", "CPN_PASSENGERS": "1", "COUPON_TYPE": "А", "CPN_DEST_ARPT_NM": "10397", "CPN_MKT_ORIG_CITY_NM": "33214", "CPN_DEST_CNTRY": "США", "CPN_MKT_ID": "201724503", "CPN_ORIG_CNTRY": "США" }, {
"CPN_STATE_NM": "Пенсильвания", "CPN_DEST": "АВА", "CPN_TKT_CARR_NM": "DL", "TRIP_BREAK": "X", "CPN_MKT_ORIG_ARPT_NM": "10397", "CLASS_OF_SVC": "X", "CPN_TKT_NBR": "2017245", "CPN_DEST_CITY_MKT_NM": "30135", "CPN_DISTANCE": "692", "Seq_num": "4", "ITIN_GEO_TYPE": "2", "CPN_RPTG_CARR_NM": "9E", "COUPON_GEO_TYPE": "2", "CPN_ORIG_STATE_NM": "Грузия", "CPN_OPERG_CARR_NM": "DL", "CPN_ORIG": "АТЛ", "CPN_PASSENGERS": "1", "COUPON_TYPE": "А", "CPN_DEST_ARPT_NM": "10135", "CPN_MKT_ORIG_CITY_NM": "30397", "CPN_DEST_CNTRY": "США", "CPN_MKT_ID": "201724503", "CPN_ORIG_CNTRY": "США" } ], "MKT_ITIN_ID": "2017245", "MKT_OPERG_CARR_NM": "DL", "MKT_DEST_ARPT_NM": "10135", "MKT_ORIG_ARPT_NM": "СБ", "MKT_ITIN_GEO_TYPE": "2", "MKT_PASSENGERS": "1", "MKT_ID": "201724503", "MKT_TKT_CARR_NM": "DL" } ], "NO_OF_CPNS": "4", "ORIGIN_COUNTRY": "США", "ITIN_ID": "2017245", "Пассажиры": "1", "MILES_FLOWN": "3132" }

Answer 1

Вы можете использовать вспомогательную функцию from_json () в вызове API набора данных select () для извлечения или декодирования атрибутов и значений данных из строки JSON в DataFrame в виде столбцов, продиктованных схемой.

пример, приведенный ниже json { "reporting_carrier": "A", "market": { "value": 10 } }, сохраненный в rawJsonDf

case class MarketData (reporting_carrier: String, market_json: String)

val jsonSchema = new StructType()
  .add("value", LongType)

rawJsonDf
  .toDf("reporting_carrier","market")
  .as[MarketData]

rawJsonDf
  .select(from_json($"market_json", jsonSchema) as "market")
  .filter($"market.value" > 5)

см. этот замечательный учебник по кирпичам для получения дополнительной информации .