Python：如何使用 Apache Beam 连接到 Snowflake？

Question

我看到 BigQuery 有一个内置的 I/O 连接器，但我们的很多数据都存储在 Snowflake 中。是否有连接到 Snowflake 的解决方法？我唯一能想到的就是用sqlalchemy运行查询，然后将输出转储到Cloud Storage Buckets，然后Apache-Beam就可以从存储在Bucket中的文件中获取输入数据。

Answer 1

Google云支持在这里！

没有从 Snowflake 到 Cloud Dataflow 的直接连接器，但您提到的解决方法是一种解决方法。首先将输出转储到 Cloud Storage，然后将 Cloud Storage 连接到 Cloud Dataflow。

希望对您有所帮助。

Answer 2

最近向 Beam 添加了 Snowflake Python 和 Java 连接器。

目前（版本 2.24）它仅支持 apache_beam.io.external.snowflake 中的 ReadFromSnowflake 操作。

在 2.25 版本中，WriteToSnowflake 也将在 apache_beam.io.snowflake 模块中可用。您仍然可以使用旧路径，但是在此版本中它将被视为已弃用。

目前它仅在 Flink Runner 上运行，但正在努力使其也可用于其他运行器。

此外，它是一个 cross-language 转换，因此可能需要一些额外的设置 - 它在此处的 pydoc 中有很好的记录（我将其粘贴在下面）： https://github.com/apache/beam/blob/release-2.24.0/sdks/python/apache_beam/io/external/snowflake.py

Snowflake transforms tested against Flink portable runner.
  **Setup**
  Transforms provided in this module are cross-language transforms
  implemented in the Beam Java SDK. During the pipeline construction, Python SDK
  will connect to a Java expansion service to expand these transforms.
  To facilitate this, a small amount of setup is needed before using these
  transforms in a Beam Python pipeline.
  There are several ways to setup cross-language Snowflake transforms.
  * Option 1: use the default expansion service
  * Option 2: specify a custom expansion service
  See below for details regarding each of these options.
  *Option 1: Use the default expansion service*
  This is the recommended and easiest setup option for using Python Snowflake
  transforms.This option requires following pre-requisites
  before running the Beam pipeline.
  * Install Java runtime in the computer from where the pipeline is constructed
    and make sure that 'java' command is available.
  In this option, Python SDK will either download (for released Beam version) or
  build (when running from a Beam Git clone) a expansion service jar and use
  that to expand transforms. Currently Snowflake transforms use the
  'beam-sdks-java-io-expansion-service' jar for this purpose.
  *Option 2: specify a custom expansion service*
  In this option, you startup your own expansion service and provide that as
  a parameter when using the transforms provided in this module.
  This option requires following pre-requisites before running the Beam
  pipeline.
  * Startup your own expansion service.
  * Update your pipeline to provide the expansion service address when
    initiating Snowflake transforms provided in this module.
  Flink Users can use the built-in Expansion Service of the Flink Runner's
  Job Server. If you start Flink's Job Server, the expansion service will be
  started on port 8097. For a different address, please set the
  expansion_service parameter.
  **More information**
  For more information regarding cross-language transforms see:
  - https://beam.apache.org/roadmap/portability/
  For more information specific to Flink runner see:
  - https://beam.apache.org/documentation/runners/flink/

Snowflake（与大多数便携式 IOs 一样）有自己的 java 扩展服务，当您没有指定自己的自定义扩展服务时，应该会自动下载该服务。我不认为应该需要它，但我提到它只是为了安全起见。您可以下载 jar 并使用 java -jar <PATH_TO_JAR> <PORT> 启动它，然后将其作为 expansion_service='localhost:<PORT>' 传递给 snowflake.ReadFromSnowflake。 Link 到 2.24 版本：https://mvnrepository.com/artifact/org.apache.beam/beam-sdks-java-io-snowflake-expansion-service/2.24.0

请注意，它仍处于实验阶段，请随时报告有关 Beam Jira 的问题。

Answer 3

对于未来寻找如何开始使用 Snowflake 和 Apache Beam 的教程的人，我可以推荐以下由连接器的创建者制作的教程。

https://www.polidea.com/blog/snowflake-and-apache-beam-on-google-dataflow/

Python：如何使用 Apache Beam 连接到 Snowflake？

Python: How to Connect to Snowflake Using Apache Beam?

python

pipeline

google-cloud-dataflow

snowflake-cloud-data-platform

apache-beam