apache-spark - 使用 Sedona Pyspark Spatial Join 广播问题

Question

因此，我将一个小型（550 行、region_name 和几何）加入到大型数据集（数百万）中，以执行 ST_Intersect 将区域分配给每个点。

我第一次运行它时它一直在运行（只有 300kish 点的大型数据集的样本）并且从未完成。我认为这很可能导致数据偏斜，并且在进行连接时广播较小的数据集是个好主意。所以我使用下面的代码。

final_df = spark.sql(
    """
    
    WITH t1 AS (
        SELECT name, ST_GeomFromWKT(geometry) as geometry
        FROM bq_provinces
    ),

    t2 AS(
        SELECT name, ST_MakeValid(geometry, false) as geometry
        FROM t1
    )
        
    SELECT /*+ MAPJOIN(t2) */ a.*, t2.name as province
    FROM raw_table as a, t2 
    WHERE ST_Intersects(ST_Point(Lon_of_Visit, Lat_of_Visit), ST_GeomFromWKT(t2.geometry))
    """
)

我做了一个 make valid 只是为了确保多边形是干净的。但是，当我添加广播时，我总是得到 Scala 匹配类型错误，如下所示

Traceback (most recent call last):
  File "/tmp/job-edaf4590/spark_script.py", line 158, in <module>
    final_df.show()
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/dataframe.py", line 484, in show
  File "/usr/lib/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py", line 1304, in __call__
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 111, in deco
  File "/usr/lib/spark/python/lib/py4j-0.10.9-src.zip/py4j/protocol.py", line 326, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o104.showString.
: scala.MatchError: SpatialIndex st_geomfromwkt(geometry#91), QUADTREE, [id=#55]
+- Generate  **org.apache.spark.sql.sedona_sql.expressions.ST_MakeValid$** , [name#84], false, [geometry#91]
   +- Project [name#84, st_geomfromwkt(geometry#85) AS geometry#89]
      +- Scan BigQueryRelation(entrada-client-242822.UberMedia_Canada_All.Canadian_Provinces
numRows=13
numBytes=81,350,432
) [name#84,geometry#85] PushedFilters: [], ReadSchema: struct<name:string,geometry:string>
 (of class org.apache.spark.sql.sedona_sql.strategy.join.SpatialIndexExec)
    at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.newQueryStage(AdaptiveSparkPlanExec.scala:457)
    at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.createQueryStages(AdaptiveSparkPlanExec.scala:416)
    at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.$anonfun$createQueryStages$2(AdaptiveSparkPlanExec.scala:446)
    at scala.collection.immutable.List.map(List.scala:297)
    at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.createQueryStages(AdaptiveSparkPlanExec.scala:446)
    at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.$anonfun$createQueryStages$2(AdaptiveSparkPlanExec.scala:446)
    at scala.collection.immutable.List.map(List.scala:293)
    at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.createQueryStages(AdaptiveSparkPlanExec.scala:446)
    at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.$anonfun$createQueryStages$2(AdaptiveSparkPlanExec.scala:446)
    at scala.collection.immutable.List.map(List.scala:293)
    at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.createQueryStages(AdaptiveSparkPlanExec.scala:446)
    at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.$anonfun$getFinalPhysicalPlan$1(AdaptiveSparkPlanExec.scala:182)
    at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
    at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.getFinalPhysicalPlan(AdaptiveSparkPlanExec.scala:179)
    at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.executeCollect(AdaptiveSparkPlanExec.scala:277)
    at org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:3696)
    at org.apache.spark.sql.Dataset.$anonfun$head$1(Dataset.scala:2722)
    at org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:3687)
    at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:103)
    at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:163)
    at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:90)
    at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
    at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
    at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3685)
    at org.apache.spark.sql.Dataset.head(Dataset.scala:2722)
    at org.apache.spark.sql.Dataset.take(Dataset.scala:2929)
    at org.apache.spark.sql.Dataset.getRows(Dataset.scala:301)
    at org.apache.spark.sql.Dataset.showString(Dataset.scala:338)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:282)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:748)

我正在使用 Spark 3.1 和 Apache Sedona。任何帮助将不胜感激

标签： apache-sparkgoogle-cloud-platformpyspark