topic Re: Getting error while loading parquet data into Postgres (using spark-postgres library) ClassNotFoundException: Failed to find data source: postgres. Please find packages at http://spark.apache.org/third-party-projects.html Caused by: ClassNotFoundException in Data Engineering

Getting error while loading parquet data into Postgres (using spark-postgres library) ClassNotFoundException: Failed to find data source: postgres. Please find packages at http://spark.apache.org/third-party-projects.html Caused by: ClassNotFoundException

explorer — Wed, 11 Jan 2023 12:10:40 GMT

Hi Fellas - I'm trying to load parquet data (in GCS location) into Postgres DB (google cloud) . For bulk upload data into PG we are using (spark-postgres library)

https://framagit.org/interhop/library/spark-etl/-/tree/master/spark-postgres/src/main/scala/io/frama/parisni/spark/postgres.

Pre requisite- To use above library , I have uploaded jar in my cluster

1- https://mvnrepository.com/artifact/io.frama.parisni/spark-postgres/0.0.1

2- https://mvnrepository.com/artifact/org.postgresql/postgresql

%scala

val data = spark.read.format("postgres")

.option("url","jdbc:postgres://IP:PORT/DATABASE?user=USERNAME&currentSchema=SCHEMANAME")

.option("password","PASSWORD")

.option("query","select * from TABLENAME")

.option("partitions",4)

.option("numSplits",5)

.option("multiline",true)

.load

But it is giving error -

ClassNotFoundException: Failed to find data source: postgres. Please find packages at http://spark.apache.org/third-party-projects.html Caused by: ClassNotFoundException: postgres.DefaultSource

P.S- Since my data is huge I need to use bulk load operation since by jdbc connection and inserting via cursor batch is not optimal method .

Let me know if there are any other methods available for bulk insert .

Data size ~ 40 GB parquet

Regards,

Re: Getting error while loading parquet data into Postgres (using spark-postgres library) ClassNotFoundException: Failed to find data source: postgres. Please find packages at http://spark.apache.org/third-party-projects.html Caused by: ClassNotFoundException

daniel_sahal — Wed, 11 Jan 2023 12:47:52 GMT

Just a lucky guess here... Maybe the spark-postgres library was not correctly installed? Try removing the jar and uploading it once again.

If that doesn't work, try to replace spark.read.format("postgres")

with spark.read.format("io.frama.parisni.spark.postgres.DataSource")

Re: Getting error while loading parquet data into Postgres (using spark-postgres library) ClassNotFoundException: Failed to find data source: postgres. Please find packages at http://spark.apache.org/third-party-projects.html Caused by: ClassNotFoundException

explorer — Mon, 16 Jan 2023 16:53:33 GMT

Hi @Kaniz Fatma and @Daniel Sahal , Thanks for reaching out , appreciate it.

Background to help you understand the context .

Problem statement-

We are facing performance issues during Postgres data loading from GCS storage (parquet format) . Data load is done in 2 parts.

First to create postgres intermediate_table by reading GCS data as spark dataframe and save it in Postgres as table .

Then from intermediate load into main table (with some logic ) .We have already implemented the code and it is working fine for smaller data sets

But when it comes to large dataset , it is posing issue .

We ensured there is no constraint in Postgres intermediate table also there is one PK in postgres main table for merging functionality. Postgres is also in GCP.

Elaborating what has been said above with code snippet.

Stage-1 ->

Creation of Postgres intermediate table (persistent table) by writing spark data frame as table in Postgres (overwrite mode) from the data stored in GCS (parquet data format ~ volume 25-30 GB, total number of part files - 71)

Code snippet –

df.write.mode("overwrite").option("numPartitions", number ).format("jdbc").option("url", url).option("dbtable", dbtable).option("user", user_nm).option("password",passwrd).save()

Stage -2 -> Inserting Postgres intermediate table (created in step-1) data into main Postgres table (persistent) via insert statement (SCD type-1).

Code

snippet –

INSERT INTO main_tab (SELECT columns FROM intermediate_table ) ON CONFLICT do something DO UPDATE SET

Postgres and Databricks resource details

Attached screen shots

Tried options ,

We ensured to use maxfilepertrigger ,batchsize , num of partition options as well in above code to ingest bulk data.

Then we saw spark-postgres option and tried below ways

option -1 – writing dataframe and giving option as “postgres” ex- write.format("postgres")

failed with - ClassNotFoundException: Failed to find data source: postgres. Please find packages

Option 2- We were trying low level pyspark API as suggested in spark-postgres-library ( ex- sc._jvm.io.frama.parisni.PGUtil) but

PGutil is not available in code as well

We tried with postgressql.driver as well but it didn’t work out

https://docs.databricks.com/external-data/postgresql.html

Regards,

Re: Getting error while loading parquet data into Postgres (using spark-postgres library) ClassNotFoundException: Failed to find data source: postgres. Please find packages at http://spark.apache.org/third-party-projects.html Caused by: ClassNotFoundException

daniel_sahal — Tue, 17 Jan 2023 08:28:59 GMT

@Deepak U

It's hard to find an answer here without the proper monitoring of the entire process.

First of all, I would check how the workers behave during runtime (Ganglia - what's the overall usage of the workers, where's the bottleneck. It could be even more efficient to have a smaller number of more powerful nodes than multiple less powerful ones.

Second thing - networking. Maybe there's a bottleneck somewhere in there? Worth checking.

Third thing - Postgres instance - monitoring during the runtime.

Fourth thing - worth considering - exporting the data to Parquet files into storage, then using COPY FROM on Postgres to import the data. It's usually faster than using JDBC.

Re: Getting error while loading parquet data into Postgres (using spark-postgres library) ClassNotFoundException: Failed to find data source: postgres. Please find packages at http://spark.apache.org/third-party-projects.html Caused by: ClassNotFoundException

explorer — Wed, 18 Jan 2023 15:44:11 GMT

Hi @Kaniz Fatma , @Daniel Sahal -

Few updates from my side.

After so many hits and trials , psycopg2 worked out in my case.

We can process 200+GB data with 10 node cluster (n2-highmem-4,32 GB Memory, 4 Cores) and driver 32 GB Memory, 4 Cores with Runtime10.4.x-scala2.12

It took close to 100 min to load whole data into PG .

Though we need to change parquet to csv and read csv sequentially to load into PG

like the below code snippet -

import psycopg2

# Create a connection to the PostgreSQL database

con = psycopg2.connect(database="dbname",user="user-nm",password="password",host="ip",port="5432")

# Create a cursor object

cur = con.cursor()

# Open the CSV file

files_to_load = dbutils.fs.ls("dbfs:/dir/")

while files_to_load:

file_path = files_to_load.pop(0).path

if file_path.endswith('.csv'):

file_path = file_path.replace("dbfs:","/dbfs")

print(file_path)

with open(file_path, "r") as f:

# Use the copy_from function to load the data into the PostgreSQL table

cur.copy_from(f, "pg_table_name", sep=",")

# Commit the changes

con.commit()

# Close the cursor and connection

con.close()

Few questions if you can assist

1- Is there any way we can tweak the code to read parquet data instead .csv data in above code (we have data originally in parquet format (databricks delta table) and if we can use same native format , we can cut down extra processing )

2- Is there any way we can run parallelly instead sequentially .

3- Any other tips which can boost the performance and take less time .

Regards,