Package org.apache.beam.sdk.io.snowflake

Class SnowflakeIO

java.lang.Object

org.apache.beam.sdk.io.snowflake.SnowflakeIO

public class SnowflakeIO
extends Object

IO to read and write data on Snowflake.

SnowflakeIO uses Snowflake JDBC driver under the hood, but data isn't read/written using JDBC directly. Instead, SnowflakeIO uses dedicated COPY operations to read/write data from/to a cloud bucket. By now only Google Cloud Storage is supported.

To configure SnowflakeIO to read/write from your Snowflake instance, you have to provide a SnowflakeIO.DataSourceConfiguration using SnowflakeIO.DataSourceConfiguration.create(). Additionally one of SnowflakeIO.DataSourceConfiguration.withServerName(String) or SnowflakeIO.DataSourceConfiguration.withUrl(String) must be used to tell SnowflakeIO which instance to use.
There are also other options available to configure connection to Snowflake:

• SnowflakeIO.DataSourceConfiguration.withWarehouse(String) to specify which Warehouse to use
• SnowflakeIO.DataSourceConfiguration.withDatabase(String) to specify which Database to connect to
• SnowflakeIO.DataSourceConfiguration.withSchema(String) to specify which schema to use
• SnowflakeIO.DataSourceConfiguration.withRole(String) to specify which role to use
• SnowflakeIO.DataSourceConfiguration.withLoginTimeout(Integer) to specify the timeout for the login
• SnowflakeIO.DataSourceConfiguration.withPortNumber(Integer) to specify custom port of Snowflake instance

For example:

 SnowflakeIO.DataSourceConfiguration dataSourceConfiguration =
     SnowflakeIO.DataSourceConfiguration.create()
         .withUsernamePasswordAuth(username, password)
         .withServerName(options.getServerName())
         .withWarehouse(options.getWarehouse())
         .withDatabase(options.getDatabase())
         .withSchema(options.getSchema());
 

Reading from Snowflake

SnowflakeIO.Read returns a bounded collection of T as a PCollection<T>. T is the type returned by the provided SnowflakeIO.CsvMapper.

For example

 PCollection<GenericRecord> items = pipeline.apply(
  SnowflakeIO.<GenericRecord>read()
    .withDataSourceConfiguration(dataSourceConfiguration)
    .fromQuery(QUERY)
    .withStagingBucketName(...)
    .withStorageIntegrationName(...)
    .withCsvMapper(...)
    .withCoder(...));
 

Important When reading data from Snowflake, temporary CSV files are created on the specified stagingBucketName in directory named `sf_copy_csv_[RANDOM CHARS]_[TIMESTAMP]`. This directory and all the files are cleaned up automatically by default, but in case of failed pipeline they may remain and will have to be cleaned up manually.

Writing to Snowflake

SnowflakeIO.Write supports writing records into a database. It writes a PCollection to the database by converting each T into a Object via a user-provided SnowflakeIO.UserDataMapper.

For example

 items.apply(
     SnowflakeIO.<KV<Integer, String>>write()
         .withDataSourceConfiguration(dataSourceConfiguration)
         .withStagingBucketName(...)
         .withStorageIntegrationName(...)
         .withUserDataMapper(maper)
         .to(table);
 

Important When writing data to Snowflake, firstly data will be saved as CSV files on specified stagingBucketName in directory named 'data' and then into Snowflake.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	SnowflakeIO.Concatenate	Combines list of String to provide one String with paths where files were staged for write.
static interface	SnowflakeIO.CsvMapper<T>	Interface for user-defined function mapping parts of CSV line into T.
static class	SnowflakeIO.DataSourceConfiguration	A POJO describing a DataSource, providing all properties allowing to create a DataSource.
static class	SnowflakeIO.DataSourceProviderFromDataSourceConfiguration	Wraps SnowflakeIO.DataSourceConfiguration to provide DataSource.
static class	SnowflakeIO.Read<T>	Implementation of read().
static interface	SnowflakeIO.UserDataMapper<T>	Interface for user-defined function mapping T into array of Objects.
static class	SnowflakeIO.Write<T>	Implementation of write().

Constructor Summary

Constructors

Constructor	Description
SnowflakeIO()

Method Summary

Modifier and Type	Method	Description
static <T> SnowflakeIO.Read<T>	read()	Read data from Snowflake via COPY statement using default SnowflakeBatchServiceImpl.
static <T> SnowflakeIO.Read<T>	read(SnowflakeServices snowflakeServices)	Read data from Snowflake via COPY statement using user-defined SnowflakeServices.
static <T> SnowflakeIO.Write<T>	write()	Write data to Snowflake via COPY statement.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

SnowflakeIO

public SnowflakeIO()

Method Details

read

public static <T> SnowflakeIO.Read<T> read(SnowflakeServices snowflakeServices)

Read data from Snowflake via COPY statement using user-defined SnowflakeServices.

Type Parameters:

T - Type of the data to be read.

Parameters:

snowflakeServices - user-defined SnowflakeServices

read

public static <T> SnowflakeIO.Read<T> read()

Read data from Snowflake via COPY statement using default SnowflakeBatchServiceImpl.

Type Parameters:

T - Type of the data to be read.

write

public static <T> SnowflakeIO.Write<T> write()

Write data to Snowflake via COPY statement.

Type Parameters:

T - Type of data to be written.