Sedona 3.5 incompatible with cloudera jar versions?

[duhizjame]

hi everyone!

I am getting an error when running a spark pipeline with sedona 1.7.2(also with 1.8.0 and 1.6.1) on Spark 3.5.4.3.5.7191000.100-2(cloudera environment):

java.lang.NoSuchMethodError: org.apache.spark.sql.execution.datasources.parquet.ParquetFooterReader.readFooter(Lorg/apache/hadoop/conf/Configuration;Lorg/apache/spark/sql/execution/datasources/PartitionedFile;Z)Lorg/apache/parquet/hadoop/metadata/ParquetMetadata;
	at org.apache.spark.sql.execution.datasources.parquet.GeoParquetFileFormat.$anonfun$buildReaderWithPartitionValues$2(GeoParquetFileFormat.scala:240)
	at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.org$apache$spark$sql$execution$datasources$FileScanRDD$$anon$$readCurrentFile(FileScanRDD.scala:219)
	at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.nextIterator(FileScanRDD.scala:282)
	at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:131)
	at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
	at org.apache.spark.sql.execution.WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.hasNext(WholeStageCodegenEvaluatorFactory.scala:43)
	at org.apache.spark.sql.execution.UnsafeExternalRowSorter.sort(UnsafeExternalRowSorter.java:225)
	at org.apache.spark.sql.execution.exchange.ShuffleExchangeExec$.$anonfun$prepareShuffleDependency$10(ShuffleExchangeExec.scala:377)
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2(RDD.scala:922)
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2$adapted(RDD.scala:922)

I have inspected the stack trace from:

      val footerFileMetaData =
        ParquetFooterReader
          .readFooter(sharedConf, file, ParquetFooterReader.SKIP_ROW_GROUPS)
          .getFileMetaData

the only difference that I see between 3.5.4 spark and cloudera flavor of the ParquetFooterReader is that this part is missing from the cloudera flavour, which matches the signature of the exception:

public class ParquetFooterReader {

  public static final boolean SKIP_ROW_GROUPS = true;
  public static final boolean WITH_ROW_GROUPS = false;

  /**
   * Reads footer for the input Parquet file 'split'. If 'skipRowGroup' is true,
   * this will skip reading the Parquet row group metadata.
   *
   * @param file a part (i.e. "block") of a single file that should be read
   * @param configuration hadoop configuration of file
   * @param skipRowGroup If true, skip reading row groups;
   *                     if false, read row groups according to the file split range
   */
  public static ParquetMetadata readFooter(
      Configuration configuration,
      PartitionedFile file,
      boolean skipRowGroup) throws IOException {
    long fileStart = file.start();
    ParquetMetadataConverter.MetadataFilter filter;
    if (skipRowGroup) {
      filter = ParquetMetadataConverter.SKIP_ROW_GROUPS;
    } else {
      filter = HadoopReadOptions.builder(configuration, file.toPath())
          .withRange(fileStart, fileStart + file.length())
          .build()
          .getMetadataFilter();
    }
    return readFooter(configuration, file.toPath(), filter);
  }

Has anyone encountered this issue on Cloudera environment? without shading this specific class, does any version work with CDS spark 3.5?
is there a way to use the unshaded sedona jars without conflict on CDP?

[jiayuasu]

https://sedona.apache.org/latest/setup/maven-coordinates/

We have both shaded and unshaded jars. Is this what you need?

[duhizjame]

hi @jiayuasu , both don't work sadly 😄

The issue is more with the Cloudera distribution; as the Sedona GeoParquetFileFormat references the method with this signature(in both shaded/unshaded variants):

public static ParquetMetadata readFooter(
      Configuration configuration,
      PartitionedFile file,
      boolean skipRowGroup

which does not exist in the cloudera distribution.

I solved the issue by manually patching the spark ParquetFooterReader and resolving the classpaths, so that it loads first on the cluster.

It would be maybe good to mention in the Sedona docs that any Sedona 3.5 version(1.5-1.8) will be incompatible with the Cloudera distribution, without manual patching. The issue is more on their side, since they should guarantee binary compatibility between their distribution and base spark, but as we can see there are some methods missing.

I leave here the updated ParquetFooterReader for reference:
(to be used with --conf spark.driver.userClassPathFirst=true --conf spark.executor.userClassPathFirst; beware of other conflicts you might create with other dependencies)

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.spark.sql.execution.datasources.parquet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.HadoopReadOptions;
import org.apache.parquet.ParquetReadOptions;
import org.apache.parquet.format.converter.ParquetMetadataConverter;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
import org.apache.parquet.hadoop.util.HadoopInputFile;
import org.apache.spark.sql.execution.datasources.PartitionedFile;

import java.io.IOException;

/**
 * `ParquetFooterReader` is a util class which encapsulates the helper
 * methods of reading parquet file footer
 */
public class ParquetFooterReader {

    public static final boolean SKIP_ROW_GROUPS = true;
    public static final boolean WITH_ROW_GROUPS = false;

    /**
     * Reads footer for the input Parquet file 'split'. If 'skipRowGroup' is true,
     * this will skip reading the Parquet row group metadata.
     *
     * @param file a part (i.e. "block") of a single file that should be read
     * @param configuration hadoop configuration of file
     * @param skipRowGroup If true, skip reading row groups;
     *                     if false, read row groups according to the file split range
     */
    public static ParquetMetadata readFooter(
            Configuration configuration,
            PartitionedFile file,
            boolean skipRowGroup) throws IOException {
        long fileStart = file.start();
        ParquetMetadataConverter.MetadataFilter filter;
        if (skipRowGroup) {
            filter = ParquetMetadataConverter.SKIP_ROW_GROUPS;
        } else {
            filter = HadoopReadOptions.builder(configuration)
                    .withRange(fileStart, fileStart + file.length())
                    .build()
                    .getMetadataFilter();
        }
        return readFooter(configuration, file.toPath(), filter);
    }

    public static ParquetMetadata readFooter(Configuration configuration,
                                             Path file, ParquetMetadataConverter.MetadataFilter filter) throws IOException {
        return readFooter(HadoopInputFile.fromPath(file, configuration), filter);
    }

    public static ParquetMetadata readFooter(Configuration configuration,
                                             FileStatus fileStatus, ParquetMetadataConverter.MetadataFilter filter) throws IOException {
        return readFooter(HadoopInputFile.fromStatus(fileStatus, configuration), filter);
    }

    private static ParquetMetadata readFooter(HadoopInputFile inputFile,
                                              ParquetMetadataConverter.MetadataFilter filter) throws IOException {
        ParquetReadOptions readOptions =
                HadoopReadOptions.builder(inputFile.getConfiguration()).withMetadataFilter(filter).build();
        // Use try-with-resources to ensure fd is closed.
        try (ParquetFileReader fileReader = ParquetFileReader.open(inputFile, readOptions)) {
            return fileReader.getFooter();
        }
    }
}

[jiayuasu]

We do have a cloud vendor tester repo here: https://github.com/wherobots/sedona-cloud-vendor-tester/tree/master

Wherobots is sponsoring the project to run Sedona integration tests on vendor specific Spark distribution. Does it make sense to add Cloudera Spark there?