Python:如何使用Apache光束连接到Snowflake？

Question

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

Answer 1

最近添加了Snowflake Python和Java连接器到横梁。

目前(2.24版)它只支持apache_beam.io.external.snowflake中的ReadFromSnowflake操作。

在2.25版本中，apache_beam.io.snowflake模块中也将提供WriteToSnowflake。您仍然可以使用旧路径，但是在此版本中它将被认为是不推荐使用的。

目前它只在Flink Runner上运行，但正在努力让其他跑步者也能使用它。

此外，它是一个跨语言转换，所以可能需要一些额外的设置-它在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。链接至2.24版本：https://mvnrepository.com/artifact/org.apache.beam/beam-sdks-java-io-snowflake-expansion-service/2.24.0

请注意，它仍然是试验性的，可以随时在Beam Jira上报告问题。

Answer 2

Google Cloud Support在这里！

没有从Snowflake到Cloud Dataflow的直接连接，但有一种解决方法就是您所提到的。首先将输出转储到云存储，然后将云存储连接到云数据流。

我希望这能有所帮助。

Answer 3

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

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