spark "basePath" option setting

apache-spark pyspark google-cloud-dataproc
jldupont picture jldupont · Nov 15, 2016 · Viewed 14k times · Source

When I do:

allf = spark.read.parquet("gs://bucket/folder/*")

I get:

java.lang.AssertionError: assertion failed: Conflicting directory structures detected. Suspicious paths:

... And the following message after the list of paths:

If provided paths are partition directories, please set "basePath" in the options of the data source to specify the root directory of the table. If there are multiple root directories, please load them separately and then union them.

I am new to Spark. I believe my data source is really a collection of "folders" (something like base/top_folder/year=x/month=y/*.parquet) and I would like to load all the files and transform them.

Thanks for your help!

  • UPDATE 1: I've looked at the Dataproc console and there is no way to set "options" when creating a cluster.
  • UPDATE 2: I've checked in the cluster's "cluster.properties" file and there is no such options. Could it be I must add one and reset the cluster?

Answer

Angus Davis picture Angus Davis · Nov 15, 2016

Per Spark documentation on Parquet partition discovery, I believe that changing your load statement from

allf = spark.read.parquet("gs://bucket/folder/*")

to

allf = spark.read.parquet("gs://bucket/folder")

should discover and load all parquet partitions. This is assuming that the data was written with "folder" as its base directory.

If the directory base/folder actually contains mutliple datasets, you will want to load each dataset independantly and then union them together.

Related questions

How to change dataframe column names in pyspark?

I come from pandas background and am used to reading data from CSV files into a dataframe and then simply changing the column names to something useful using the simple command: df.columns = new_column_name_list However, the same …

Read More
Load CSV file with Spark

I'm new to Spark and I'm trying to read CSV data from a file with Spark. Here's what I am doing : sc.textFile('file.csv') .map(lambda line: (line.split(',')[0], line.split(',')[1])) .collect() I would expect …

Read More
How do I add a new column to a Spark DataFrame (using PySpark)?

I have a Spark DataFrame (using PySpark 1.5.1) and would like to add a new column. I've tried the following without any success: type(randomed_hours) # => list # Create in Python and transform to RDD new_col = pd.DataFrame(randomed_hours, …

Read More