Hi Jia,
Thanks again for your advice. I am still running into performance issues and
pursuing a slightly different approach of converting points into a MULTIPOINT
geometry instead in order to reduce the number of spatial joins.
This is still running very slowly using the dataframe api, so I am trying with
spatial RDDs.
I am running into a parsing issue on a nested array defined as:
val joinSchema = StructType(Seq(
StructField("leftgeometry", StringType, true),
StructField("trip_id", StringType, true),
StructField("probe_id", StringType, true),
StructField("provider_id", StringType, true),
StructField("mode", IntegerType, true),
...
StructField("points", ArrayType(StructType(Seq(
StructField("capture_time", StringType, true),
StructField("lat", DoubleType, true),
StructField("lon", DoubleType, true),
StructField("raw_speed", DoubleType, true)
)), true), true),
StructField("rightgeometry",
...
|-- points: array (nullable = true)
| |-- element: struct (containsNull = true)
| | |-- capture_time: string (nullable = true)
| | |-- lat: double (nullable = true)
| | |-- lon: double (nullable = true)
| | |-- raw_speed: double (nullable = true)
|-- rightgeometry: string (nullable = true)
WrappedArray([2022-03-03 23:19:32.000,40.971663,-124.10638,NaN], [2022-03-03
23:27:31.000,40.972371,-124.104135,107.568], [2022-03-03
23:28:40.000,41.002168,-124.143558,325.990692]
Error:
023-01-10 22:39:33 ERROR Executor:94 - Exception in task 0.0 in stage 10.0 (TID
13)
scala.MatchError:
ArrayType(StructType(StructField(capture_time,StringType,true),
StructField(lat,DoubleType,true), StructField(lon,DoubleType,true),
StructField(raw_speed,DoubleType,true)),true) (of class
org.apache.spark.sql.types.ArrayType)
at org.apache.sedona.sql.utils.Adapter$.parseString(Adapter.scala:301)
at
org.apache.sedona.sql.utils.Adapter$.$anonfun$castRowToSchema$1(Adapter.scala:271)
at
scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:273)
at scala.collection.immutable.List.foreach(List.scala:392)
at scala.collection.TraversableLike.map(TraversableLike.scala:273)
at scala.collection.TraversableLike.map$(TraversableLike.scala:266)
at scala.collection.immutable.List.map(List.scala:298)
at
org.apache.sedona.sql.utils.Adapter$.castRowToSchema(Adapter.scala:268)
at
org.apache.sedona.sql.utils.Adapter$.$anonfun$toDf$7(Adapter.scala:189)
at scala.collection.Iterator$$anon$10.next(Iterator.scala:461)
at scala.collection.Iterator$$anon$10.next(Iterator.scala:461)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown
Source)
at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
Does Sedona support the parsing of nested arrays? Also is there planned support
for maps?
Thanks,
Trang
-----Original Message-----
From: Jia Yu <[email protected]>
Sent: Sunday, January 8, 2023 10:21 PM
To: [email protected]
Subject: Re: Propagating user defined attributes to spatial join
Use Good Judgement: This email originated outside of INRIX Do not click on
links or open attachments unless you recognize the sender and know the content
is safe.
Hi Trang,
1. Since Sedona 1.3.1-incubating, you can add user defined struct type in the
join result (pair RDD) to recover the original types. See:
https://nam04.safelinks.protection.outlook.com/?url=https%3A%2F%2Fsedona.apache.org%2F1.3.1-incubating%2Ftutorial%2Fsql%2F%23spatialpairrdd-to-dataframe&data=05%7C01%7CTrang.Nguyen%40inrix.com%7Cde1832975e5644e589ba08daf209b715%7C6ad2e4da8c924e588877ed06b8918379%7C0%7C0%7C638088420806028486%7CUnknown%7CTWFpbGZsb3d8eyJWIjoiMC4wLjAwMDAiLCJQIjoiV2luMzIiLCJBTiI6Ik1haWwiLCJXVCI6Mn0%3D%7C3000%7C%7C%7C&sdata=OH1JqGsxCD1U29T4i%2FJU15%2FWFBcTI8UxIaNiQmtRIrk%3D&reserved=0
Example 1:
val schema = StructType(Array( StructField("county", GeometryUDT, nullable =
true), StructField("name", StringType, nullable = true), StructField("price"
, DoubleType, nullable = true), StructField("age", IntegerType, nullable =
true) )) val spatialDf = Adapter.toDf(spatialRDD, schema, sparkSession)
Example 2:
val schema = StructType(Array(
StructField("leftGeometry", GeometryUDT, nullable = true),
StructField("name", StringType, nullable = true),
StructField("price", DoubleType, nullable = true),
StructField("age", IntegerType, nullable = true),
StructField("rightGeometry", GeometryUDT, nullable = true),
StructField("category", StringType, nullable = true)
))
val joinResultDf = Adapter.toDf(joinResultPairRDD, schema, sparkSession)
2. For DataFrame SQL join,
(1) since you are using ST_Intersects, please try to swap two operands in
ST_Intersects. E.g., ST_Intersects (Trips, Points) to ST_Intersects(Points,
Trips). This does not affect the correctness of the join result.
(2) Your query has lots of filters on non-spatial attributes and this can
actually reduce the join query overhead. Usually Sedona should be able to do
filter pushdown then run the join. But if the query plan does not work that way
(check it by printing joinDf.explain()), you can force Spark to it:
Run all the filters in SQL query 1, cache the result. Then run the spatial join
in SQL query 2.
For example,
tripDf = spark.sql(SELECT XXX FROM XXX WHERE trip_distance_m < 1000.0 and
provider_id in ('1000')).cache()
tripDf.createOrReplaceTempView("tripDf")
joinDf = spark.sql(SELECT XXX FROM tripDf, pointDf WHERE ST_Intersects(trip,
points))
Thanks,
Jia
On Sun, Jan 8, 2023 at 1:03 PM Trang Nguyen <[email protected]> wrote:
> Hi Jia,
>
>
>
> Thanks a lot for the hints. Reversing the left and right field names
> did take care of the issue.
>
> However all the datatypes for the original fields were converted into
> StringType. Is there a way to preserve the original type mapping as well?
>
>
>
> 0 = {StructField@21788} StructField(leftgeometry,GeometryUDT,true)
>
> 1 = {StructField@21789} StructField(poi_id,StringType,true)
>
> 2 = {StructField@21790} StructField(poi_parent_id,StringType,true)
>
> 3 = {StructField@21791} StructField(lat,StringType,true)
>
> 4 = {StructField@21792} StructField(lon,StringType,true)
>
> ....
>
> 22 = {StructField@21810} StructField(probe_id,StringType,true)
>
> 23 = {StructField@21811} StructField(trip_distance_m,StringType,true)
>
> 24 = {StructField@21812} StructField(trip_id,StringType,true)
>
> 25 = {StructField@21813} StructField(capture_time,StringType,true)
>
>
>
> --
>
>
>
> For the dataframe based implementation, you are right that
> repartitioning the source dataframes did increase the spatial partition.
>
> However, the performance is incredibly slow. It is taking over 2 hours
> to perform a spatial st_intersects join between two tables (poi, 9GB
> and exploded points, 50GB).
>
> The query I am running is:
>
> with explodedPoints
>
> as
>
> (select is_moving, end_time, start_time, provider_id, mode,
> probe_id, trip_distance_m, trip_id, size(points) as point_count,
> explode(points) as point from t)
>
> select distinct is_moving, end_time, start_time,
> provider_id, mode, probe_id, trip_distance_m, trip_id, point.lat,
> point.lon, point.capture_time , poi.poi_id, poi.qk, poi.poi_parent_id,
> poi.lat as poi_lat, poi.lon as poi_lon, poi.owned, poi.fleetcentric,
> to_date(opened_on) as opened_on, to_date(opened_on) as closed_on from
> explodedPoints
>
> join poi where trip_distance_m < 1000.0 and provider_id in
> ('1000') and st_intersects(geom, ST_Point(point.lon, point.lat)) and
> point_count >= 2 and point_count <= 500
>
>
>
>
>
> DAG:
>
>
>
>
>
>
>
>
>
> Do you see any way to speed up the performance?
>
>
> Thanks,
> Trang
>
> -----Original Message-----
> From: Jia Yu <[email protected]>
> Sent: Saturday, January 7, 2023 10:38 PM
> To: [email protected]
> Subject: Re: Propagating user defined attributes to spatial join
>
>
>
> Use Good Judgement: This email originated outside of INRIX Do not
> click on links or open attachments unless you recognize the sender and
> know the content is safe.
>
>
>
> Hi Trang,
>
>
>
> 1. For Sedona SQL join, you usually don't need to set other parameters
> via conf. The repartition will simply work. The new num partitions
> might be preserved in the final result and it is something Sedona
> tried to achieve but not guaranteed.
>
> 2. For RDD Join, as shown in the two links I gave you:
>
>
>
> var spatialRDD = Adapter.toSpatialRdd(spatialDf, "usacounty")
>
>
>
> This will bring field names to SpatialRDD (saved in
> SpatialRDD.fieldNames attribute).
>
>
>
> When you finish your spatial join on two SpatialRDD and get joinResult
> (which is a pairRdd), run.
>
>
>
> import scala.collection.JavaConversions._
>
> var joinResultDf = Adapter.toDf(joinResultPairRDD, leftRdd.fieldNames,
> rightRdd.fieldNames, sparkSession)
>
>
>
> This will give the dataframe with the original column names. If it
> doesn't work, try to swap the fieldNames:
> Adapter.toDf(joinResultPairRDD, rightRdd.fieldNames,
> leftRdd.fieldNames, sparkSession)
>
>
>
>
>
> On Sat, Jan 7, 2023 at 11:30 PM Trang Nguyen <[email protected]>
> wrote:
>
>
>
> > Hi Jia,
>
> >
>
> > Thanks for the quick response. I also tried to repartition trips as
>
> > below to different values but never see the count I specified during
>
> > the spatial join itself.
>
> > Not exactly sure why but I am trying now against the spatial RDDs
> directly.
>
> > Is there a way to propagate the custom fields across?
>
> >
>
> > Thanks
>
> > Trang
>
> >
>
> > -----Original Message-----
>
> > From: Jia Yu <[email protected]>
>
> > Sent: Saturday, January 7, 2023 10:26 PM
>
> > To: [email protected]
>
> > Subject: Re: Propagating user defined attributes to spatial join
>
> >
>
> > Use Good Judgement: This email originated outside of INRIX Do not
>
> > click on links or open attachments unless you recognize the sender
> > and
>
> > know the content is safe.
>
> >
>
> > Hi Trang,
>
> >
>
> > The slow join performance issue is mostly caused by too few partitions.
>
> >
>
> > The most straightforward way to increase the number of partitions is:
>
> > repartition both input DataFrame right after you load them from disk.
>
> >
>
> > e.g.,
>
> >
>
> > var tripDf = spark.read(XXX)
>
> > tripDf = tripDf.repartition(tripDf.numPartitions * 5)
>
> >
>
> > var poiDf = spark.read(XXX)
>
> > poiDf = poiDf.repartition(poiDf.numPartitions * 5)
>
> >
>
> > Then perform the SQL spatial join
>
> > ====
>
> >
>
> > If you want to use RDD API,
>
> >
>
> > Please read [1] and [2]
>
> >
>
> > [1]
>
> >
>
> > https://nam04.safelinks.protection.outlook.com/?url=https%3A%2F%2Fse
> > do%2F&data=05%7C01%7CTrang.Nguyen%40inrix.com%7Cde1832975e5644e589ba
> > 08daf209b715%7C6ad2e4da8c924e588877ed06b8918379%7C0%7C0%7C6380884208
> > 06028486%7CUnknown%7CTWFpbGZsb3d8eyJWIjoiMC4wLjAwMDAiLCJQIjoiV2luMzI
> > iLCJBTiI6Ik1haWwiLCJXVCI6Mn0%3D%7C3000%7C%7C%7C&sdata=aZgP2BVZZ6DVRO
> > FEnGangkJGSJ1LH6VOf5H1hYxuEv4%3D&reserved=0
>
> > na.apache.org%2F1.3.1-incubating%2Ftutorial%2Fsql%2F%23dataframe-to-
> > sp
>
> > atialrdd&data=05%7C01%7CTrang.Nguyen%40inrix.com%7C56aca1c2bca3495be
> > 34
>
> > a08daf142ffc1%7C6ad2e4da8c924e588877ed06b8918379%7C0%7C0%7C638087567
> > 31
>
> > 1005763%7CUnknown%7CTWFpbGZsb3d8eyJWIjoiMC4wLjAwMDAiLCJQIjoiV2luMzIi
> > LC
>
> > JBTiI6Ik1haWwiLCJXVCI6Mn0%3D%7C3000%7C%7C%7C&sdata=zhJ3muIkE40wFmOTH
> > wd
>
> > mWNlEorBQq5neN9taEFEvw3I%3D&reserved=0
>
> > [2]
>
> >
>
> > https://nam04.safelinks.protection.outlook.com/?url=https%3A%2F%2Fse
> > do%2F&data=05%7C01%7CTrang.Nguyen%40inrix.com%7Cde1832975e5644e589ba
> > 08daf209b715%7C6ad2e4da8c924e588877ed06b8918379%7C0%7C0%7C6380884208
> > 06028486%7CUnknown%7CTWFpbGZsb3d8eyJWIjoiMC4wLjAwMDAiLCJQIjoiV2luMzI
> > iLCJBTiI6Ik1haWwiLCJXVCI6Mn0%3D%7C3000%7C%7C%7C&sdata=aZgP2BVZZ6DVRO
> > FEnGangkJGSJ1LH6VOf5H1hYxuEv4%3D&reserved=0
>
> > na.apache.org%2F1.3.1-incubating%2Ftutorial%2Fsql%2F%23spatialpairrd
> > d-
>
> > to-dataframe&data=05%7C01%7CTrang.Nguyen%40inrix.com%7C56aca1c2bca34
> > 95
>
> > be34a08daf142ffc1%7C6ad2e4da8c924e588877ed06b8918379%7C0%7C0%7C63808
> > 75
>
> > 67311005763%7CUnknown%7CTWFpbGZsb3d8eyJWIjoiMC4wLjAwMDAiLCJQIjoiV2lu
> > Mz
>
> > IiLCJBTiI6Ik1haWwiLCJXVCI6Mn0%3D%7C3000%7C%7C%7C&sdata=SlL8QXYxXU1yd
> > C5
>
> > KFggp2HB7E%2BF0bY0Kw5dmBvwr644%3D&reserved=0
>
> >
>
> > Thanks,
>
> > Jia
>
> >
>
> >
>
> >
>
> > On Sat, Jan 7, 2023 at 11:14 PM Trang Nguyen
> > <[email protected]>
>
> > wrote:
>
> >
>
> > > Hi,
>
> > >
>
> > > I'm newbie to Sedona and am running into performance issues using
>
> > > the dataframe api to handle spatial joins with st_intersects.
> > > From
>
> > > the execution plan, I see that this is largely due to too few
>
> > > partitions getting used for the spatial join.
>
> > >
>
> > > I've tried to increase the partition size by setting:
>
> > > sparkSession.conf.set("spark.sql.shuffle.partitions",
>
> > > appConf.getInputPartitions.toString)
>
> > > sparkSession.conf.set("geospark.join.numpartition",
>
> > > appConf.getInputPartitions.toString)
>
> > > sparkSession.conf.set("spark.default.parallelism",
>
> > > appConf.getInputPartitions.toString)
>
> > >
>
> > > The setting changse seem to make minimal difference.
>
> > >
>
> > > I'm trying now to use convert the dataframes to be joined into
>
> > > spatialRDDs so that I can set the number of partitions for the
>
> > > spatial
>
> > join.
>
> > > However, I am running into a different issue when I try to convert
>
> > > back from the spatial joined result into a dataframe because the
>
> > > extra attributes from the original dataframes are not getting
>
> > > propagated through the join.
>
> > >
>
> > > I am using the Adapter class for the conversion.
>
> > >
>
> > >
>
> > > val tripRDD = Adapter.toSpatialRdd(trips, "geom", tripColumns) val
>
> > > rddWithOtherAttributes = tripRDD.rawSpatialRDD.rdd.map[String](f=>
> > > {
>
> > > f.getUserData.toString
>
> > > })
>
> > > tripRDD.analyze()
>
> > >
>
> > > val poiRDD = Adapter.toSpatialRdd(poiDS.toDF, "geom", poiColumns)
>
> > > poiRDD.analyze()
>
> > >
>
> > >
>
> > >
>
> > > tripRDD.spatialPartitioning(GridType.KDBTREE,
>
> > > appConf.getInputPartitions)
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > > val joinRes = JoinQuery.SpatialJoinQueryFlat(tripRDD, poiRDD,
>
> > > usingIndex,
>
> > > spatialPredicate)
>
> > >
>
> > > val df = Adapter.toDf(joinRes, tripColumns, poiColumns,
>
> > > sparkSession)
>
> > >
>
> > > df.show
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > > How can I get the original attributes to be propagated as part of
>
> > > the join? I searched the documentation but couldn't find any
>
> > > documentation on this.
>
> > > By specifying the columns to be carried through in the
>
> > > Adapter.toSpatialRdd, I assume that the attributes would be
> > > carried
>
> > > through into the join as well.
>
> > >
>
> > > Here is the error I am seeing:
>
> > > va.lang.RuntimeException: Error while encoding:
>
> > > java.lang.RuntimeException: java.lang.String is not a valid
> > > external
>
> > > type for schema of geometry if (assertnotnull(input[0,
>
> > > org.apache.spark.sql.Row, true]).isNullAt) null else
>
> > > newInstance(class
>
> > > org.apache.spark.sql.sedona_sql.UDT.GeometryUDT).serialize AS
>
> > > leftgeometry#2424
>
> > > if (assertnotnull(input[0, org.apache.spark.sql.Row,
>
> > > true]).isNullAt) null else staticinvoke(class
>
> > > org.apache.spark.unsafe.types.UTF8String,
>
> > > StringType, fromString,
>
> > > validateexternaltype(getexternalrowfield(assertnotnull(input[0,
>
> > > org.apache.spark.sql.Row, true]), 1, leftgeometry), StringType),
>
> > > true, false, true) AS leftgeometry#2425 if (assertnotnull(input[0,
>
> > > org.apache.spark.sql.Row, true]).isNullAt) null else
>
> > > staticinvoke(class org.apache.spark.unsafe.types.UTF8String,
>
> > > StringType, fromString,
>
> > > validateexternaltype(getexternalrowfield(assertnotnull(input[0,
>
> > > org.apache.spark.sql.Row, true]), 2, trip_id), StringType), true,
>
> > > false,
>
> > > true) AS trip_id#2426
>
> > > if (assertnotnull(input[0, org.apache.spark.sql.Row,
>
> > > true]).isNullAt) null else newInstance(class
>
> > > org.apache.spark.sql.sedona_sql.UDT.GeometryUDT).serialize AS
>
> > > rightgeometry#2427
>
> > > at
>
> > >
>
> > org.apache.spark.sql.errors.QueryExecutionErrors$.expressionEncoding
> > Er
>
> > ror(QueryExecutionErrors.scala:1052)
>
> > > at
>
> > >
>
> > org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$Serializer.
> > ap
>
> > ply(ExpressionEncoder.scala:210)
>
> > > at
>
> > > org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$Serializer.
>
> > > ap
>
> > > ply(ExpressionEncoder.scala:193)
>
> > >
>
> > > Thanks,
>
> > > Trang
>
> > >
>
> >
>
