Snowflake

Snowflake 是一款全托管的 SaaS（软件即服务），提供一个统一平台用于数据仓库、数据湖、数据工程、数据科学、数据应用开发以及实时/共享数据的安全共享与使用。Snowflake 开箱即用的功能包括存储与计算分离、按需弹性计算、数据共享、数据克隆以及对第三方工具的支持，以满足日益增长的企业的苛刻需求。

先决条件

您需要一个已启动并运行的 Snowflake 实例。如果还没有，可以在这里创建。

依赖

所需的依赖项有

net.snowflake:spark-snowflake_<scala_version>:<version>
net.snowflake:snowflake-jdbc:<version>

从 Snowflake 到 Neo4j

// Step (1)
// Load a table into a Spark DataFrame
val snowflakeDF: DataFrame = spark.read
  .format("snowflake")
  .option("sfURL", "<account_identifier>.snowflakecomputing.com")
  .option("sfUser", "<user_name>")
  .option("sfPassword", "<password>")
  .option("sfDatabase", "<database>")
  .option("sfSchema", "<schema>")
  .option("dbtable", "CUSTOMER")
  .load()

// Step (2)
// Save the `snowflakeDF` as nodes with labels `Person` and `Customer` into Neo4j
snowflakeDF.write
  .format("org.neo4j.spark.DataSource")
  .mode(SaveMode.ErrorIfExists)
  .option("url", "neo4j://<host>:<port>")
  .option("labels", ":Person:Customer")
  .save()

# Step (1)
# Load a table into a Spark DataFrame
snowflakeDF = (spark.read
  .format("snowflake")
  .option("sfURL", "<account_identifier>.snowflakecomputing.com")
  .option("sfUser", "<user_name>")
  .option("sfPassword", "<password>")
  .option("sfDatabase", "<database>")
  .option("sfSchema", "<schema>")
  .option("dbtable", "CUSTOMER")
  .load())

# Step (2)
# Save the `snowflakeDF` as nodes with labels `Person` and `Customer` into Neo4j
(snowflakeDF.write
  .format("org.neo4j.spark.DataSource")
  .mode(SaveMode.ErrorIfExists)
  .option("url", "neo4j://<host>:<port>")
  .option("labels", ":Person:Customer")
  .save())

从 Neo4j 到 Snowflake

// Step (1)
// Load `:Person:Customer` nodes as DataFrame
val neo4jDF: DataFrame = spark.read.format("org.neo4j.spark.DataSource")
  .option("url", "neo4j://<host>:<port>")
  .option("labels", ":Person:Customer")
  .load()

// Step (2)
// Save the `neo4jDF` as table CUSTOMER into Snowflake
neo4jDF.write
  .format("snowflake")
  .mode("overwrite")
  .option("sfURL", "<account_identifier>.snowflakecomputing.com")
  .option("sfUser", "<user_name>")
  .option("sfPassword", "<password>")
  .option("sfDatabase", "<database>")
  .option("sfSchema", "<schema>")
  .option("dbtable", "CUSTOMER")
  .save()

# Step (1)
# Load `:Person:Customer` nodes as DataFrame
neo4jDF = (spark.read.format("org.neo4j.spark.DataSource")
  .option("url", "neo4j://<host>:<port>")
  .option("labels", ":Person:Customer")
  .load())

# Step (2)
# Save the `neo4jDF` as table CUSTOMER into Snowflake
(neo4jDF.write
  .format("snowflake")
  .mode("overwrite")
  .option("sfURL", "<account_identifier>.snowflakecomputing.com")
  .option("sfUser", "<user_name>")
  .option("sfPassword", "<password>")
  .option("sfDatabase", "<database>")
  .option("sfSchema", "<schema>")
  .option("dbtable", "CUSTOMER")
  .save())