Error I'm getting when I write Pyspark code to connect with Snowflake

Viewed 1559

I'm getting an error when I tried to write PySpark code from Jupiter Notebook to connect with Snowflake. Here's the error I got:

Py4JJavaError: An error occurred while calling o526.load. : java.lang.ClassNotFoundException: Failed to find data source: net.snowflake.spark.snowflake. Please find packages at http://spark.apache.org/third-party-projects.html

Spark-version: v2.4.5 Master: local[*] Python 3.X

Here's my code:

from pyspark import SparkConf, SparkContext
from pyspark.sql import SQLContext
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

sc = SparkContext.getOrCreate()

spark = SparkSession.builder \
    .master("local") \
    .appName("Test") \
    .config('spark.jars','/Users/zhao/Downloads/snowflake-jdbc-3.5.4.jar,/Users/zhao/Downloads/spark-snowflake_2.11-2.3.2.jar') \
    .getOrCreate()

sfOptions = {
  "sfURL" : "xxx",
  "sfUser" : "xxx",
  "sfPassword" : "xxx",
  "sfDatabase" : "xxx",
  "sfSchema" : "xxx",
  "sfWarehouse" : "xxx",
  "sfRole": "xxx"
}

SNOWFLAKE_SOURCE_NAME = "net.snowflake.spark.snowflake"

df = spark.read.format(SNOWFLAKE_SOURCE_NAME) \
  .options(**sfOptions) \
  .option("query",  "select * from CustomerInfo limit 10") \
  .load()

I'll appreciate if anyone can give me some ideas :)

4 Answers

How are you starting up your jupyter notebook server instance? Are you ensuring your PYTHONPATH and SPARK_HOME variables are properly set, and that Spark isn't pre-running an instance? Also, is your Snowflake Spark Connector jar using the right Spark and Scala version variants?

Here's a fully bootstrapped and tested run on a macOS machine, as a reference (uses homebrew):

# Install JDK8
~> brew tap adoptopenjdk/openjdk
~> brew cask install adoptopenjdk8

# Install Apache Spark (v2.4.5 as of post date)
~> brew install apache-spark

# Install Jupyter Notebooks (incl. optional CLI notebooks)
~> pip3 install --user jupyter notebook

# Ensure we use JDK8 (using very recent JDKs will cause class version issues)
~> export JAVA_HOME="/Library/Java/JavaVirtualMachines/adoptopenjdk-8.jdk/Contents/Home"

# Setup environment to allow discovery of PySpark libs and the Spark binaries
# (Uses homebrew features to set the paths dynamically)
~> export SPARK_HOME="$(brew --prefix apache-spark)/libexec"
~> export PYTHONPATH="${SPARK_HOME}/python:${SPARK_HOME}/python/build:${PYTHONPATH}"
~> export PYTHONPATH="$(brew list apache-spark | grep 'py4j-.*-src.zip$' | head -1):${PYTHONPATH}"

# Download jars for dependencies in notebook code into /tmp

# Snowflake JDBC (v3.12.8 used here):
~> curl --silent --location \
'https://search.maven.org/classic/remotecontent?filepath=net/snowflake/snowflake-jdbc/3.12.8/snowflake-jdbc-3.12.8.jar' \
> /tmp/snowflake-jdbc-3.12.8.jar

# Snowflake Spark Connector (v2.7.2 used here)
# But more importantly, a Scala 2.11 and Spark 2.4.x compliant one is fetched
~> curl --silent --location \
'https://search.maven.org/classic/remotecontent?filepath=net/snowflake/spark-snowflake_2.11/2.7.2-spark_2.4/spark-snowflake_2.11-2.7.2-spark_2.4.jar' \
> /tmp/spark-snowflake_2.11-2.7.2-spark_2.4.jar

# Run the jupyter notebook service (opens up in webbrowser)
~> jupyter notebook

Code run within a new Python 3 notebook:

from pyspark import SparkConf, SparkContext
from pyspark.sql import SQLContext
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

sfOptions = {
    "sfURL": "account.region.snowflakecomputing.com",
    "sfUser": "username",
    "sfPassword": "password",
    "sfDatabase": "db_name",
    "sfSchema": "schema_name",
    "sfWarehouse": "warehouse_name",
    "sfRole": "role_name",
}

spark = SparkSession.builder \
    .master("local") \
    .appName("Test") \
    .config('spark.jars','/tmp/snowflake-jdbc-3.12.8.jar,/tmp/spark-snowflake_2.11-2.7.2-spark_2.4.jar') \
    .getOrCreate()

SNOWFLAKE_SOURCE_NAME = "net.snowflake.spark.snowflake"

df = spark.read.format(SNOWFLAKE_SOURCE_NAME) \
    .options(**sfOptions) \
    .option("query",  "select * from CustomerInfo limit 10") \
    .load()

df.show()

The above example uses a read method (to move data from Snowflake into Spark) but if you'd like to write a dataframe instead, see the documentation on moving data from Spark to Snowflake.

I have followed the steps described in the answer from @user13472370, with the same library versions, Additionally, I am using the same snowflake connection parameters to connect to Snowflave from SQL Workbench. However I am still receiving the same error

An error occurred while calling o43.load.
    : java.lang.NoClassDefFoundError: scala/Product$class
        at net.snowflake.spark.snowflake.Parameters$MergedParameters.<init>(Parameters.scala:288)

Update: Meanwhile I found an easy to implement solution using the AWS Glue service: https://aws.amazon.com/blogs/big-data/performing-data-transformations-using-snowflake-and-aws-glue/

Probably a bit late, but another alternative for future users: in my case this issue was due to version incompatibilities, after lots of trial and error, and seeing examples on the web, I found a snowflake jdbc, spark connector and python versions compatible with our(old) spark cluster version(2.4.4), no code nor credential changes, and this error was gone.

Related