Skip to content

Support for Snowflake connector with Spark #3364

New issue
New issue
Closed
Closed
Support for Snowflake connector with Spark#3364
Labels
kind/featureNew feature or requestwontfixThis will not be worked on

Description

@amithadiraju1694
amithadiraju1694
opened on Nov 28, 2022

Is your feature request related to a problem? Please describe.

For folks on data bricks or other spark-based platforms, getting large data as a pandas df may not be an option .... It's possible to get a spark df using Snowflake Connector for Python but it's orders of magnitude slower compared to Snowflake connector for Spark

Describe the solution you'd like
Add a function to SnowflakeRetrievalJob which supports executing & retrieving snowflake query through Snowflake Connector for Spark

Describe alternatives you've considered
Tried getting snowflake query results as arrow_batches -> converting to pandas on each batch -> to spark data frame -> concatenate all spark dos to one. This is orders go marinated slower than reading same data frame from Snowflake Connector for Spark.

cc: @sfc-gh-madkins

Activity

  1. amithadiraju1694
    added
    kind/featureNew feature or request
    on Nov 28, 2022
  2. adchia
    closed this as completedon Dec 15, 2022

  3. amithadiraju1694 commented on Dec 15, 2022

    @amithadiraju1694
    amithadiraju1694
    on Dec 15, 2022
    ContributorAuthor

    @adchia would love to use this if completed, since this is one of major pain points for us. Could you pls provide link for docs on how to use this functionality?

  4. adchia
    reopened this on Dec 15, 2022

  5. adchia commented on Dec 15, 2022

    @adchia
    adchia
    on Dec 15, 2022
    Collaborator

    Oh woops I had misread this one. Yeah the SparkOfflineStore doesn't connect to Snowflake yet. Feel free to contribute this and I can cut it in a new patch release!

    The easiest way to add this

    That being said, it should be a matter of adding another mode in the spark_source.py file.

    With logic looking something like below being in the get_table_query_string:

    options = {
    "sfUrl": url,
    "sfUser": user,
    "sfPassword": password,
    "sfDatabase": database,
    "sfSchema": schema,
    "sfWarehouse": warehouse,
    "APPLICATION": "feast",
}

if role:
    options["sfRole"] = role

df_reader = spark.read.format("snowflake").options(**options)

if table:
    df_reader = df_reader.option("dbtable", table)
else:
    df_reader = df_reader.option("query", query)

df = df_reader.load()
tmp_table_name = get_temp_entity_table_name()
df.createOrReplaceTempView(tmp_table_name)

    I'd probably add something like an optional snowflake_config in SparkSource to capture the above configs (e.g. warehouse, database, etc)

    The better way to add this

    Really though, we should support reading SnowflakeSource files and mapping that into the above.

    This would probably live in https://github.com/feast-dev/feast/blob/master/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py#L76, and instead of failing on SparkSource, also support parsing SnowflakeSource. And then constructing the table_query_string like above.

  6. sfc-gh-madkins
    added
    priority/p2
    priority/p1
    and removed
    priority/p2
    on Dec 19, 2022

  7. amithadiraju1694 commented on Dec 20, 2022

    @amithadiraju1694
    amithadiraju1694
    on Dec 20, 2022
    ContributorAuthor

    Oh woops I had misread this one. Yeah the SparkOfflineStore doesn't connect to Snowflake yet. Feel free to contribute this and I can cut it in a new patch release!

    The easiest way to add this

    That being said, it should be a matter of adding another mode in the spark_source.py file.

    With logic looking something like below being in the get_table_query_string:

    options = {
    "sfUrl": url,
    "sfUser": user,
    "sfPassword": password,
    "sfDatabase": database,
    "sfSchema": schema,
    "sfWarehouse": warehouse,
    "APPLICATION": "feast",
}

if role:
    options["sfRole"] = role

df_reader = spark.read.format("snowflake").options(**options)

if table:
    df_reader = df_reader.option("dbtable", table)
else:
    df_reader = df_reader.option("query", query)

df = df_reader.load()
tmp_table_name = get_temp_entity_table_name()
df.createOrReplaceTempView(tmp_table_name)

    I'd probably add something like an optional snowflake_config in SparkSource to capture the above configs (e.g. warehouse, database, etc)

    The better way to add this

    Really though, we should support reading SnowflakeSource files and mapping that into the above.

    This would probably live in https://github.com/feast-dev/feast/blob/master/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py#L76, and instead of failing on SparkSource, also support parsing SnowflakeSource. And then constructing the table_query_string like above.

    Thanks for getting back ! Few questions based on your response:

    1. If this functionality ( reading from snowflake source ) is added in SparkOfflineStore , aren't we necessarily creating additional data in hive meta-store ?

    2. Along same lines, can we add this functionality to SnowflakeOfflineStore, by adding additional functions to SnowflakeRetrievalJob to use Snowflake Connector for Spark instead of Snowflake Connector for Python ?

    My thought's that, doing through 2 one can still use snowflake as offline store without essentially creating a additional copy of data elsewhere. Let me know your thoughts.

  8. sfc-gh-madkins commented on Dec 27, 2022

    @sfc-gh-madkins
    sfc-gh-madkins
    on Dec 27, 2022
    Collaborator

    @amithadiraju1694 this should be possible to do

  9. sfc-gh-madkins commented on Dec 27, 2022

    @sfc-gh-madkins
    sfc-gh-madkins
    on Dec 27, 2022
    Collaborator

    @amithadiraju1694 this is the alternative you tried correct but found it slow? #3358

  10. amithadiraju1694 commented on Dec 28, 2022

    @amithadiraju1694
    amithadiraju1694
    on Dec 28, 2022
    ContributorAuthor

    @amithadiraju1694 this is the alternative you tried correct but found it slow? #3358

    Yup, that's the one I tried.

  11. amithadiraju1694 commented on Dec 28, 2022

    @amithadiraju1694
    amithadiraju1694
    on Dec 28, 2022
    ContributorAuthor

    @amithadiraju1694 this should be possible to do

    I tried doing it in SnowflakeRetrievalJob , by sending connection options to spark and using spark connector. It couldn't find feast_df_***** object , and I wasn't sure how to fix it. So I stopped there. Please let me know if you have a solution for that.

  12. sfc-gh-madkins commented on Dec 28, 2022

    @sfc-gh-madkins
    sfc-gh-madkins
    on Dec 28, 2022 via email
    Collaborator
  13. sfc-gh-madkins
    removed
    priority/p1
    on Apr 22, 2023

  14. sfc-gh-madkins commented on Apr 24, 2023

    @sfc-gh-madkins
    sfc-gh-madkins
    on Apr 24, 2023
    Collaborator

    @amithadiraju1694 are you able to give me a side by side numbers comparison using to_spark_df() vs materializing the table in snowflake and then reading it through the spark connector?

  15. stale commented on Sep 17, 2023

    @stale
    staleBot
    on Sep 17, 2023

    This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

  16. stale
    added
    wontfixThis will not be worked on
    on Sep 17, 2023
  17. stale
    closed this as completedon Mar 17, 2024
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    kind/featureNew feature or requestwontfixThis will not be worked on

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions