如何在 Scala 中使用 Spark SQL 返回多个 JSON 对象

Posted 2023-04-17

【中文标题】如何在 Scala 中使用 Spark SQL 返回多个 JSON 对象

【英文标题】：How to return multiple JSON objects using Spark SQL in Scala

【发布时间】：2017-12-12 02:43:14

【问题描述】：

我有一个 csv 文件和一个 json 文件。在 count.csv 中有三列（纬度、经度、计数）。在 json 中，这里是一个示例：

  "type": "Feature",
  "properties": 
  "ID": "15280000000231",
  "TYPES": "Second Class",
  "N2C": "9",
  "NAME": "Century Road"
  ,
  "geometry": 
    "type": "LineString",
    "coordinates": [
      [
        6.1395489,
        52.3107973
      ],
      [
        6.1401178,
        52.3088457
      ],
      [
        6.1401126,
        52.3088071
      ]
    ]
  

目前我的 scala 代码近似并匹配经度和纬度，并过滤 csv 文件以匹配 lon/lat，返回 lat/lon 并计为 csv。

我想从 json 中返回所有属性（ID、TYPE、N2C 和 NAME），并将匹配的纬度/经度作为原始线串 + csv 中的计数 + json 中的属性全部作为 json 文件而不是.csv。

到目前为止，我一直在努力做到这一点？

case class ScoredLocation(latitude: Double, longitude: Double, score: Int)

object ScoreFilter 
  val Epsilon = 10000

  val DoubleToRoundInt = udf(
    (coord:Double) => (coord * Epsilon).toInt
  )

  val schema = Encoders.product[ScoredLocation].schema
  val route_count = spark.read.schema(schema).format("csv").load("count.csv")
    .withColumn("lat_aprx", DoubleToRoundInt($"latitude"))
    .withColumn("lon_aprx", DoubleToRoundInt($"longitude"))

  val match_route = spark.read.format("json").load("matchroute.json")
    .select(explode($"geometry.coordinates"))
    .select($"col".getItem(0).alias("latitude"), $"col".getItem(1).alias("longitude"))
    .withColumn("lat_aprx", DoubleToRoundInt($"latitude"))
    .withColumn("lon_aprx", DoubleToRoundInt($"longitude"))

  europe_count.show()
  scenic_route.show()

  val result = route_count.join(match_route, Seq("lat_aprx", "lon_aprx"), "leftsemi")
    .select($"latitude", $"longitude", $"count")

  result.show()
  result.write.format("csv").save("result.csv")

回复编辑：

我在使用解决方案时遇到此错误。

Exception in thread "main" org.apache.spark.sql.AnalysisException: 
cannot resolve '`ID`' given input columns: [count, latitude, 
longitude, lat_aprx, lon_aprx];;
'Project [latitude#3, longitude#4, score#5, 'ID, 'TYPES, 'N2C, 'NAME]
+- Project [lat_aprx#10, lon_aprx#16, latitude#3, longitude#4,score#5]
  +- Join LeftSemi, ((lat_aprx#10 = lat_aprx#55) && (lon_aprx#16 = lon_aprx#63))
     :- Project [latitude#3, longitude#4, score#5, lat_aprx#10, if 
(isnull(longitude#4)) null else UDF(longitude#4) AS lon_aprx#16]
      :  +- Project [latitude#3, longitude#4, count#5, if 
(isnull(latitude#3)) null else UDF(latitude#3) AS lat_aprx#10]
      :     +- Relation[latitude#3,longitude#4,count#5] csv
      +- Project [ID#38, TYPES#39, N2C#40, NAME#41, coords#48, 
lat_aprx#55, if (isnull(coords#48[1])) null else UDF(coords#48[1]) AS 
lon_aprx#63]
         +- Project [ID#38, TYPES#39, N2C#40, NAME#41, coords#48, if 
 (isnull(coords#48[0])) null else UDF(coords#48[0]) AS lat_aprx#55]
             +- Project [properties#32.ID AS ID#38, 
properties#32.TYPES AS TYPES#39, properties#32.N2C AS N2C#40, 
properties#32.NAME AS NAME#41, coords#48]
                +- Generate explode(geometry#31.coordinates), true, 
 false, [coords#48]
                  +- Relation[geometry#31,properties#32,type#33] json

编辑 2：我现在返回 json 并添加了计数，但现在的问题是返回原始 geojson，作为 linestring 类型，并添加计数，示例如下。它应该更像上面的原始json。我想它可以在之后进行操作，但我希望将其作为一个 spark sql 进程来执行。有什么想法吗？

  
   "lat":5.2509524,
   "lon":53.3926721,
   "count":1,
   "ID":"15280000814947",
   "TYPES":"Second Class",
   "N2C":"9"
  
   "lat":5.251464,
   "lon":53.3919782,
   "count":4,
   "ID":"15280000814947",
   "TYPES":"Second Class",
   "N2C":"9"
  
   "lat":5.251674,
   "lon":53.3916119,
   "count":4,
   "ID":"15280000814947",
   "TYPES":"Second Class",
   "N2C":"9"

【参考方案1】：

当您处理 match_route 数据框时，请确保选择您实际想要保留的所有列。例如：

val match_route = spark.read.format("json").load("matchroute.json")
  .select($"properties.*", explode($"geometry.coordinates").as("coords"))
  .withColumn("latitude", $"coords".getItem(0))
  .withColumn("longitude", $"coords".getItem(1))
  .withColumn("lat_aprx", DoubleToRoundInt(latitude))
  .withColumn("lon_aprx", DoubleToRoundInt(longitude))
  .drop($"coords")

确保在最后一次选择中也添加相关列，

val result = route_count.join(match_route, Seq("lat_aprx", "lon_aprx"), "leftsemi")
  .select($"latitude", $"longitude", $"count", $"ID", $"TYPES", $"N2C", $"NAME")

【讨论】：

感谢您的回复。我不确定如何正确添加编辑响应，我在加入时遇到错误，现在最后选择。我在我原来的帖子上面添加了它。其次，这个返回的纬度/经度不会是原始的线串？

@WhaleSaver 我更新了答案，现在看起来好多了。是的，我错过了最后选择的 latitude 和 longitude 列，将它们添加到数据框中。

它最终成为了问题所在的leftsemi，但我已经通过内部连接解决了这个问题。我编辑了原件以显示它现在返回的内容。理想情况下，我想要 linestring geojson，如果可能的话，有什么想法吗？

@WhaleSaver：我不熟悉 linestring geojson，但我认为应该可以。我建议就它提出一个新问题，因为它看起来会成为一个很好的单独问题。