From 3f2fbced5a216c8ad9ed4b0ebb6cf68fe746f152 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?S=C5=82awomir=20Andrian?= <>
Date: Wed, 8 Jul 2020 16:18:06 +0200
Subject: [PATCH 1/5] Added SnowflakeIO connector guide

---
 .../en/documentation/io/built-in/snowflake.md | 738 ++++++++++++++++++
 1 file changed, 738 insertions(+)
 create mode 100644 website/www/site/content/en/documentation/io/built-in/snowflake.md

diff --git a/website/www/site/content/en/documentation/io/built-in/snowflake.md b/website/www/site/content/en/documentation/io/built-in/snowflake.md
new file mode 100644
index 000000000000..f8024c044c20
--- /dev/null
+++ b/website/www/site/content/en/documentation/io/built-in/snowflake.md
@@ -0,0 +1,738 @@
+---
+title: "Apache Snowflake I/O connector"
+---
+<!--
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+
+[Built-in I/O Transforms](/documentation/io/built-in/)
+
+# Snowflake I/O
+Pipeline options and general information about using and running Snowflake IO.
+
+## Authentication
+All authentication methods available for the Snowflake JDBC Driver are possible to use with the IO transforms:
+
+- Username and password
+- Key pair
+- OAuth token
+
+Passing credentials is done via Pipeline options.
+
+Passing credentials is done via Pipeline options used to instantiate `SnowflakeIO.DataSourceConfiguration`:
+{{< highlight java >}}
+SnowflakePipelineOptions options = PipelineOptionsFactory
+        .fromArgs(args)
+        .withValidation()
+        .as(SnowflakePipelineOptions.class);
+SnowflakeCredentials credentials = SnowflakeCredentialsFactory.of(options);
+
+SnowflakeIO.DataSourceConfiguration.create(credentials)
+        .(other DataSourceConfiguration options)
+{{< /highlight >}}
+### Username and password 
+To use username/password authentication in SnowflakeIO, invoke your pipeline with the following Pipeline options:
+{{< highlight >}}
+--username=<USERNAME> --password=<PASSWORD>
+{{< /highlight >}}
+### Key pair
+To use this authentication method, you must first generate a key pair and associate the public key with the Snowflake user that will connect using the IO transform. For instructions,  see the [Snowflake documentation](https://docs.snowflake.com/en/user-guide/jdbc-configure.html).
+
+To use key pair authentication with SnowflakeIO, invoke your pipeline with one of the following set of Pipeline options:
+{{< highlight >}}
+--username=<USERNAME> --privateKeyPath=<PATH_TO_P8_FILE> --privateKeyPassphrase=<PASSWORD_FOR_KEY>
+--username=<USERNAME> --rawPrivateKey=<PRIVATE_KEY> --privateKeyPassphrase=<PASSWORD_FOR_KEY>
+{{< /highlight >}}
+
+### OAuth token
+SnowflakeIO also supports OAuth token.
+
+**IMPORTANT**: SnowflakeIO requires a valid OAuth access token. It will neither be able to refresh the token nor obtain it using a web-based flow. For information on configuring an OAuth integration and obtaining the token, see the  [Snowflake documentation](https://docs.snowflake.com/en/user-guide/oauth-intro.html).
+
+Once you have the token, invoke your pipeline with following Pipeline Options:
+{{< highlight >}}
+--oauthToken=<TOKEN>
+{{< /highlight >}}
+## DataSource Configuration
+DataSource configuration is required in both read and write object for configuring Snowflake connection properties for IO purposes.
+### General usage
+Create the DataSource configuration:
+{{< highlight java >}}
+ SnowflakeIO.DataSourceConfiguration
+            .create()
+            .withUrl(options.getUrl())
+            .withServerName(options.getServerName())
+            .withDatabase(options.getDatabase())
+            .withWarehouse(options.getWarehouse())
+            .withSchema(options.getSchema());
+{{< /highlight >}}
+Where parameters can be:
+
+- ` .withUrl(...)`
+  - JDBC-like URL for your Snowflake account, including account name and region, without any parameters.
+  - Example: `.withUrl("jdbc:snowflake://account.snowflakecomputing.com")`
+- `.withServerName(...)`
+  - Server Name - full server name with account, zone and domain.
+  - Example: `.withServerName("account.snowflakecomputing.com")`
+- `.withDatabase(...)`
+  - Name of the Snowflake database to use. 
+  - Example: `.withDatabase("MY_DATABASE")`
+- `.withWarehouse(...)`
+  - Name of the Snowflake warehouse to use. This parameter is optional. If no warehouse name is specified, the default warehouse for the user is used.
+  - Example: `.withWarehouse("MY_WAREHOUSE")`
+- `.withSchema(...)`
+  - Name of the schema in the database to use. This parameter is optional.
+  - Example: `.withSchema("PUBLIC")`
+- `.withUsernamePasswordAuth(username, password)`
+  - Sets username/password authentication.
+  - Example: `.withUsernamePasswordAuth("USERNAME", "PASSWORD")`
+- `.withOAuth(token)`
+  - Sets OAuth authentication.
+  - Example: `.withOAuth("TOKEN")`
+- `.withKeyPairAuth(username, privateKey)`
+  - Sets key pair authentication using username and [PrivateKey](https://docs.oracle.com/javase/8/docs/api/java/security/PrivateKey.html)
+  - Example: `.withKeyPairAuth("USERNAME",` [PrivateKey](https://docs.oracle.com/javase/8/docs/api/java/security/PrivateKey.html)`)`
+- `.withKeyPairPathAuth(username, privateKeyPath, privateKeyPassphrase)`
+  - Sets key pair authentication using username, path to private key file and passphrase.
+  - Example: `.withKeyPairPathAuth("USERNAME", "PATH/TO/KEY.P8", "PASSPHRASE")`
+- `.withKeyPairRawAuth(username, rawPrivateKey, privateKeyPassphrase)`
+  - Sets key pair authentication using username, private key and passphrase.
+  - Example: `.withKeyPairRawAuth("USERNAME", "PRIVATE_KEY", "PASSPHRASE")`
+
+
+**Note** - either `.withUrl(...)` or `.withServerName(...)` **is required**.
+## Pipeline options
+Use Beam’s [Pipeline options](https://beam.apache.org/releases/javadoc/2.17.0/org/apache/beam/sdk/options/PipelineOptions.html) to set options via the command line.
+### Snowflake Pipeline options
+Snowflake IO library supports following options that can be passed via the [command line](https://beam.apache.org/documentation/io/built-in/snowflake/#running-main-command-with-pipeline-options) by default when a Pipeline uses them:
+
+`--url` Snowflake's JDBC-like url including account name and region without any parameters.
+
+`--serverName` Full server name with account, zone and domain.
+
+`--username` Required for username/password and Private Key authentication.
+
+`--oauthToken` Required for OAuth authentication only.
+
+`--password` Required for username/password authentication only.
+
+`--privateKeyPath` Path to Private Key file. Required for Private Key authentication only.
+
+`--rawPrivateKey` Private Key. Required for Private Key authentication only.
+
+`--privateKeyPassphrase` Private Key's passphrase. Required for Private Key authentication only.
+
+`--stagingBucketName` External bucket path ending with `/`. I.e. `gs://bucket/`. Sub-directories are allowed.
+
+`--storageIntegrationName` Storage integration name
+
+`--warehouse` Warehouse to use. Optional.
+
+`--database` Database name to connect to. Optional.
+
+`--schema` Schema to use. Optional.
+
+`--table` Table to use. Optional.
+
+`--query` Query to use. Optional.
+
+`--role` Role to use. Optional.
+
+`--authenticator` Authenticator to use. Optional.
+
+`--portNumber` Port number. Optional.
+
+`--loginTimeout` Login timeout. Optional.
+
+`--snowPipe` SnowPipe name. Optional.
+
+In the connector code SnowflakePipelineOptions looks following (and it can be [extended](https://beam.apache.org/documentation/io/built-in/snowflake/#extending-pipeline-options)):
+{{< highlight java >}}
+public interface SnowflakePipelineOptions extends PipelineOptions, StreamingOptions {
+  String BASIC_CONNECTION_INFO_VALIDATION_GROUP = "BASIC_CONNECTION_INFO_GROUP";
+  String AUTH_VALIDATION_GROUP = "AUTH_VALIDATION_GROUP";
+
+  @Description(
+      "Snowflake's JDBC-like url including account name and region without any parameters.")
+  @Validation.Required(groups = BASIC_CONNECTION_INFO_VALIDATION_GROUP)
+  String getUrl();
+
+  void setUrl(String url);
+
+  @Description("Server Name - full server name with account, zone and domain.")
+  @Validation.Required(groups = BASIC_CONNECTION_INFO_VALIDATION_GROUP)
+  ValueProvider<String> getServerName();
+
+  void setServerName(ValueProvider<String> serverName);
+
+  @Description("Username. Required for username/password and Private Key authentication.")
+  @Validation.Required(groups = AUTH_VALIDATION_GROUP)
+  ValueProvider<String> getUsername();
+
+  void setUsername(ValueProvider<String> username);
+
+  @Description("OAuth token. Required for OAuth authentication only.")
+  @Validation.Required(groups = AUTH_VALIDATION_GROUP)
+  String getOauthToken();
+
+  void setOauthToken(String oauthToken);
+
+  @Description("Password. Required for username/password authentication only.")
+  @Default.String("")
+  ValueProvider<String> getPassword();
+
+  void setPassword(ValueProvider<String> password);
+
+  @Description("Path to Private Key file. Required for Private Key authentication only.")
+  @Default.String("")
+  String getPrivateKeyPath();
+
+  void setPrivateKeyPath(String privateKeyPath);
+
+  @Description("Private key. Required for Private Key authentication only.")
+  @Default.String("")
+  ValueProvider<String> getRawPrivateKey();
+
+  void setRawPrivateKey(ValueProvider<String> rawPrivateKey);
+
+  @Description("Private Key's passphrase. Required for Private Key authentication only.")
+  @Default.String("")
+  ValueProvider<String> getPrivateKeyPassphrase();
+
+  void setPrivateKeyPassphrase(ValueProvider<String> keyPassphrase);
+
+  @Description("Warehouse to use. Optional.")
+  @Default.String("")
+  ValueProvider<String> getWarehouse();
+
+  void setWarehouse(ValueProvider<String> warehouse);
+
+  @Description("Database name to connect to. Optional.")
+  @Default.String("")
+  ValueProvider<String> getDatabase();
+
+  void setDatabase(ValueProvider<String> database);
+
+  @Description("Schema to use. Optional.")
+  @Default.String("")
+  ValueProvider<String> getSchema();
+
+  void setSchema(ValueProvider<String> schema);
+
+  @Description("Table to use. Optional.")
+  @Default.String("")
+  ValueProvider<String> getTable();
+
+  void setTable(ValueProvider<String> table);
+
+  @Description("Query to use. Optional.")
+  @Default.String("")
+  ValueProvider<String> getQuery();
+
+  void setQuery(ValueProvider<String> query);
+
+  @Description("Role to use. Optional.")
+  @Default.String("")
+  ValueProvider<String> getRole();
+
+  void setRole(ValueProvider<String> role);
+
+  @Description("Authenticator to use. Optional.")
+  @Default.String("")
+  String getAuthenticator();
+
+  void setAuthenticator(String authenticator);
+
+  @Description("Port number. Optional.")
+  @Default.String("")
+  String getPortNumber();
+
+  void setPortNumber(String portNumber);
+
+  @Description("Login timeout. Optional.")
+  @Default.String("")
+  String getLoginTimeout();
+
+  void setLoginTimeout(String loginTimeout);
+
+  @Description("Temporary GCS bucket name.")
+  ValueProvider<String> getStagingBucketName();
+
+  void setStagingBucketName(ValueProvider<String> stagingBucketName);
+
+  @Description("Storage integration name")
+  ValueProvider<String> getStorageIntegrationName();
+
+  void setStorageIntegrationName(ValueProvider<String> storageIntegrationName);
+
+  @Description("SnowPipe name. Optional.")
+  ValueProvider<String> getSnowPipe();
+
+  void setSnowPipe(ValueProvider<String> snowPipe);
+}
+{{< /highlight >}}
+### Using Pipeline options
+To use Pipeline options, you must configure them as follows:
+
+{{< highlight java>}}
+SnowflakePipelineOptions options = PipelineOptionsFactory
+    .fromArgs(args)
+    .withValidation()
+    .as(SnowflakePipelineOptions.class);
+{{< /highlight >}}
+
+All the below parameters are required:
+- `.fromArgs()`
+  - GNU style command line arguments, 
+  - Example: `--project=myproject --database=test`
+- `.withValidation()`
+  - which validates that PipelineOptions confirms all criteria from the passed in interface
+- `.as()`
+  - a class of used pipeline options
+  - Example: `.as(SnowflakePipelineOptions.class)`
+
+Then create your pipeline using created options:
+
+{{< highlight java >}}
+Pipeline pipeline = Pipeline.create(options);
+{{< /highlight >}}
+
+Example of accessing pipeline options in code:
+
+{{< highlight java >}}
+String stagingBucketName = options.getStagingBucketName();
+{{< /highlight >}}
+
+Pipelines in tests:
+{{< highlight java >}}
+PipelineOptionsFactory.register(SnowflakePipelineOptions.class);
+    options = TestPipeline
+                .testingPipelineOptions()                                  
+                .as(SnowflakePipelineOptions.class);
+{{< /highlight >}}
+
+Then use them in pipeline run:
+{{< highlight java >}}
+PipelineResult pipelineResult = pipeline.run(options);
+{{< /highlight >}}
+
+### Extending pipeline options
+Extend the SnowflakePipelineOptions with your own custom options to access additional parameters in your code.
+
+Example of extending the Pipeline options:
+{{< highlight java >}}
+public interface BatchTestPipelineOptions extends SnowflakePipelineOptions {
+  @Description("Table name to connect to.")
+  String getTable();
+
+  void setTable(String table);
+}
+{{< /highlight >}}
+
+The following example shows how to insert the custom options:
+{{< highlight java >}}
+ExampleSnowflakePipelineOptions options = PipelineOptionsFactory
+.fromArgs(args)
+.withValidation()
+.as(ExampleSnowflakePipelineOptions.class);
+
+options.getTable();
+{{< /highlight >}}
+
+And in case of testing:
+{{< highlight java >}}
+PipelineOptionsFactory.register(ExampleSnowflakePipelineOptions.class);
+    options = TestPipeline
+    .testingPipelineOptions()
+                .as(ExampleSnowflakePipelineOptions.class);
+{{< /highlight >}}
+
+### Running main command with Pipeline options
+To pass Pipeline options via the command line, use `--args` in a gradle command as follows:
+
+{{< highlight >}}
+./gradle run 
+    --args="
+        --serverName=<SNOWFLAKE SERVER NAME>  
+           Example: --serverName=account.region.gcp.snowflakecomputing.com
+        --username=<SNOWFLAKE USERNAME> 
+           Example: --username=testuser
+        --password=<SNOWFLAKE PASSWORD> 
+           Example: --password=mypassword
+        --database=<SNOWFLAKE DATABASE> 
+           Example: --database=TEST_DATABASE
+        --schema=<SNOWFLAKE SCHEMA> 
+           Example: --schema=public
+        --table=<SNOWFLAKE TABLE IN DATABASE> 
+           Example: --table=TEST_TABLE
+        --query=<IF NOT TABLE THEN QUERY> 
+           Example: --query=‘SELECT column FROM TABLE’
+        --storageIntegrationName=<SNOWFLAKE STORAGE INTEGRATION NAME> 
+           Example: --storageIntegrationName=my_integration
+        --stagingBucketName=<GCS BUCKET NAME>
+           Example: --stagingBucketName=gs://my_gcp_bucket/
+        --runner=<DirectRunner/DataflowRunner>
+           Example: --runner=DataflowRunner
+        --project=<FOR DATAFLOW RUNNER: GCP PROJECT NAME> 
+           Example: --project=my_project
+        --tempLocation=<FOR DATAFLOW RUNNER: GCS TEMP LOCATION STARTING
+                        WITH gs://…>
+           Example: --tempLocation=gs://my_bucket/temp/
+        --region=<FOR DATAFLOW RUNNER: GCP REGION> 
+           Example: --region=us-east-1
+        --appName=<OPTIONAL: DATAFLOW JOB NAME PREFIX>
+           Example: --appName=my_job
+    "
+    {{< /highlight >}}
+Then in the code it is possible to access the parameters with arguments using the options.getStagingBucketName(); command.
+
+### Running test command with Pipeline options
+To pass Pipeline options via the command line, use `-DintegrationTestPipelineOptions` in a gradle command as follows:
+{{< highlight >}}
+./gradlew test --tests nameOfTest 
+-DintegrationTestPipelineOptions='[
+  "--serverName=<SNOWFLAKE SERVER NAME>",
+      Example: --serverName=account.region.gcp.snowflakecomputing.com
+  "--username=<SNOWFLAKE USERNAME>",
+      Example: --username=testuser
+  "--password=<SNOWFLAKE PASSWORD>",
+      Example: --password=mypassword
+  "--schema=<SNOWFLAKE SCHEMA>",
+      Example: --schema=PUBLIC
+  "--table=<SNOWFLAKE TABLE IN DATABASE>",
+      Example: --table=TEST_TABLE
+  "--database=<SNOWFLAKE DATABASE>",
+      Example: --database=TEST_DATABASE
+  "--storageIntegrationName=<SNOWFLAKE STORAGE INTEGRATION NAME>", 
+      Example: --storageIntegrationName=my_integration
+  "--stagingBucketName=<GCS BUCKET NAME>",
+      Example: --stagingBucketName=gs://my_gcp_bucket
+  "--externalLocation=<GCS BUCKET URL STARTING WITH GS://>",
+      Example: --tempLocation=gs://my_bucket/temp/
+]' --no-build-cache
+{{< /highlight >}}
+
+Where all parameters are starting with “--”, they are surrounded with double quotation and separated with comma:
+
+- `--serverName=<SNOWFLAKE SERVER NAME>`
+  - Specifies the full name of your account (provided by Snowflake). Note that your full account name might include additional segments that identify the region and cloud platform where your account is hosted. 
+  - Example: `--serverName=xy12345.eu-west-1.gcp..snowflakecomputing.com`
+
+- `--username=<SNOWFLAKE USERNAME>`
+  - Specifies the login name of the user.
+  - Example: `--username=my_username`
+
+- `--password=<SNOWFLAKE PASSWORD>`
+  - Specifies the password for the specified user.
+  - Example: `--password=my_secret`
+
+- `--schema=<SNOWFLAKE SCHEMA>`
+  - Specifies the schema to use for the specified database once connected. The specified schema should be an existing schema for which the specified user’s role has privileges.
+  - Example: `--schema=PUBLIC`
+
+- `--table=<SNOWFLAKE TABLE IN DATABASE>`
+  - Example: `--table=MY_TABLE`
+
+- `--database=<SNOWFLAKE DATABASE>`
+  - Specifies the database to use once connected. The specified database should be an existing database for which the specified user’s role has privileges.
+  - Example: `--database=MY_DATABASE`
+
+- `--storageIntegrationName=<SNOWFLAKE STORAGE INTEGRATION NAME>`
+  - Name of storage integration created in [Snowflake](https://docs.snowflake.com/en/sql-reference/sql/create-storage-integration.html) for a cloud storage of choice.
+  - Example: `--storageIntegrationName=my_google_integration`
+
+## Running pipelines on Dataflow
+By default, pipelines are run on [Direct Runner](https://beam.apache.org/documentation/runners/direct/) on your local machine. To run a pipeline on [Google Dataflow](https://cloud.google.com/dataflow/), you must provide the following Pipeline options:
+
+- `--runner=DataflowRunner`
+  - The Dataflow’s specific runner.
+
+- `--project=<GCS PROJECT>`
+  - Name of the Google Cloud Platform project.
+
+- `--stagingBucketName=<GCS BUCKET NAME>`
+  - Google Cloud Services bucket where the Beam files will be staged.
+
+- `--maxNumWorkers=5`
+  - (optional) Maximum number of workers.
+
+- `--appName=<JOB NAME>`
+  - (optional) Prefix for the job name in the Dataflow Dashboard.
+
+More pipeline options for Dataflow can be found [here](https://beam.apache.org/releases/javadoc/2.17.0/org/apache/beam/runners/dataflow/options/DataflowPipelineOptions.html).
+
+**Note**: To properly authenticate with Google Cloud, please use [gcloud](https://cloud.google.com/sdk/gcloud/) or follow the [Google Cloud documentation](https://cloud.google.com/docs/authentication/).
+
+**Important**: Please acknowledge [Google Dataflow pricing](Important: Please acknowledge Google Dataflow pricing).
+
+###Running pipeline templates on Dataflow
+Google Dataflow is supporting [template](https://cloud.google.com/dataflow/docs/guides/templates/overview) creation which means staging pipelines on Cloud Storage and running them with ability to pass runtime parameters that are only available during pipeline execution.
+
+The process of creating own Dataflow template is following
+1. Create your own pipeline. 
+2. Create [Dataflow template](https://cloud.google.com/dataflow/docs/guides/templates/creating-templates#creating-and-staging-templates) with checking which options SnowflakeIO is supporting at runtime.
+3. Run a Dataflow template using [Cloud Console](https://cloud.google.com/dataflow/docs/guides/templates/running-templates#using-the-cloud-console), [REST API](https://cloud.google.com/dataflow/docs/guides/templates/running-templates#using-the-rest-api) or [gcloud](https://cloud.google.com/dataflow/docs/guides/templates/running-templates#using-gcloud).
+
+Currently, SnowflakeIO supports following options at runtime:
+
+- `--serverName` Full server name with account, zone and domain.
+
+- `--username` Required for username/password and Private Key authentication.
+
+- `--password` Required for username/password authentication only.
+
+- `--rawPrivateKey` Private Key file. Required for Private Key authentication only.
+
+- `--privateKeyPassphrase` Private Key's passphrase. Required for Private Key authentication only.
+
+- `--stagingBucketName` external bucket path ending with `/`. I.e. `gs://bucket/`. Sub-directories are allowed.
+
+- `--storageIntegrationName` Storage integration name.
+
+- `--warehouse` Warehouse to use. Optional.
+
+- `--database` Database name to connect to. Optional.
+
+- `--schema` Schema to use. Optional.
+
+- `--table` Table to use. Optional. Note: table is not in default pipeline options.
+
+- `--query` Query to use. Optional. Note: query is not in default pipeline options.
+
+- `--role` Role to use. Optional.
+
+- `--snowPipe` SnowPipe name. Optional.
+
+**Note**: table and query is not in pipeline options by default, it may be added by [extending](https://beam.apache.org/documentation/io/built-in/snowflake/#extending-pipeline-options) PipelineOptions.
+
+Currently, SnowflakeIO **doesn't support** following options at runtime:
+
+- `--url` Snowflake's JDBC-like url including account name and region without any parameters.
+
+- `--oauthToken` Required for OAuth authentication only.
+
+- `--privateKeyPath` Path to Private Key file. Required for Private Key authentication only.
+
+- `--authenticator` Authenticator to use. Optional.
+
+- `--portNumber` Port number. Optional.
+
+- `--loginTimeout` Login timeout. Optional.
+
+## Writing to Snowflake tables
+One of the functions of SnowflakeIO is writing to Snowflake tables. This transformation enables you to finish the Beam pipeline with an output operation that sends the user's [PCollection](https://beam.apache.org/releases/javadoc/2.17.0/org/apache/beam/sdk/values/PCollection.html) to your Snowflake database.
+### Batch write (from a bounded source)
+The basic .`write()` operation usage is as follows:
+{{< highlight java >}}
+data.apply(
+   SnowflakeIO.<type>write()
+       .withDataSourceConfiguration(dc)
+       .toTable("MY_TABLE")
+       .withStagingBucketName("BUCKET NAME")
+       .withStorageIntegrationName("STORAGE INTEGRATION NAME")
+       .withUserDataMapper(mapper)
+)
+{{< /highlight >}}
+Replace type with the data type of the PCollection object to write; for example, SnowflakeIO.<String> for an input PCollection of Strings.
+
+All the below parameters are required:
+
+- `.withDataSourceConfiguration()` Accepts a DatasourceConfiguration object.
+
+- `.toTable()` Accepts the target Snowflake table name.
+
+- `.withStagingBucketName()` Accepts a cloud bucket path ended with slash.
+ -Example: `.withStagingBucketName("gs://mybucket/my/dir/")`
+
+- `.withStorageIntegrationName()` Accepts a name of a Snowflake storage integration object created according to Snowflake documentationt. Example:
+{{< highlight >}}
+CREATE OR REPLACE STORAGE INTEGRATION test_integration 
+TYPE = EXTERNAL_STAGE 
+STORAGE_PROVIDER = GCS 
+ENABLED = TRUE 
+STORAGE_ALLOWED_LOCATIONS = ('gcs://bucket/');
+{{< /highlight >}}
+Then:
+{{< highlight >}}
+.withStorageIntegrationName(test_integration)
+{{< /highlight >}}
+
+- `.withUserDataMapper()` Accepts the UserDataMapper function that will map a user's PCollection to an array of String values `(String[])`.
+
+**Note**:
+SnowflakeIO uses COPY statements behind the scenes to write (using [COPY to table](https://docs.snowflake.net/manuals/sql-reference/sql/copy-into-table.html)). StagingBucketName will be used to save CSV files which will end up in Snowflake. Those CSV files will be saved under the “stagingBucketName” path.
+
+**Optional** for batching:
+- `.withQuotationMark()` 
+  - Default value: ‘ (single quotation mark).
+  - Accepts String with one character. It will surround all text (String) fields saved to CSV. It should be one of the accepted characters by [Snowflake’s](https://docs.snowflake.com/en/sql-reference/sql/create-file-format.html) [FIELD_OPTIONALLY_ENCLOSED_BY](https://docs.snowflake.com/en/sql-reference/sql/create-file-format.html) parameter (double quotation mark, single quotation mark or none).
+  - Example: `.withQuotationMark("'")`
+
+### UserDataMapper function
+The UserDataMapper function is required to map data from a PCollection to an array of String values before the `write()` operation saves the data to temporary .csv files. For example:
+{{< highlight java >}}
+public static SnowflakeIO.UserDataMapper<Long> getCsvMapper() {
+    return (SnowflakeIO.UserDataMapper<Long>) recordLine -> new String[] {recordLine.toString()};
+}
+{{< /highlight >}}
+
+### Additional write options
+#### Transformation query
+The `.withQueryTransformation()` option for the `write()` operation accepts a SQL query as a String value, which will be performed while transfering data staged in CSV files directly to the target Snowflake table. For information about the transformation SQL syntax,  see the [Snowflake Documentation](https://docs.snowflake.net/manuals/sql-reference/sql/copy-into-table.html#transformation-parameters).
+
+Usage:
+{{< highlight java >}}
+String query = "SELECT t.$1 from YOUR_TABLE;";
+data.apply(
+   SnowflakeIO.<~>write()
+       .withDataSourceConfiguration(dc)
+       .toTable("MY_TABLE")
+       .withStagingBucketName("BUCKET NAME")
+       .withStorageIntegrationName("STORAGE INTEGRATION NAME")
+       .withUserDataMapper(mapper)
+       .withQueryTransformation(query)
+)
+{{< /highlight >}}
+
+#### Write disposition
+Define the write behaviour based on the table where data will be written to by specifying the `.withWriteDisposition(...)` option for the `write()` operation. The following values are supported:
+
+- APPEND - Default behaviour. Written data is added to the existing rows in the table,
+
+- EMPTY - The target table must be empty;  otherwise, the write operation fails,
+
+- TRUNCATE - The write operation deletes all rows from the target table before writing to it.
+
+Example of usage:
+{{< highlight java >}}
+data.apply(
+   SnowflakeIO.<~>write()
+       .withDataSourceConfiguration(dc)
+       .toTable("MY_TABLE")
+       .withStagingBucketName("BUCKET NAME")
+       .withStorageIntegrationName("STORAGE INTEGRATION NAME")
+       .withUserDataMapper(mapper)
+       .withWriteDisposition(TRUNCATE)
+)
+{{< /highlight >}}
+
+#### Create disposition
+The `.withCreateDisposition()` option defines the behavior of the write operation if the target table does not exist . The following values are supported:
+
+- CREATE_IF_NEEDED - default behaviour. The write operation checks whether the specified target table exists; if it does not, the write operation attempts to create the table Specify the schema for the target table using the `.withTableSchema()` option.
+
+- CREATE_NEVER -  The write operation fails if the target table does not exist.
+
+Usage:
+{{< highlight java >}}
+data.apply(
+   SnowflakeIO.<~>write()
+       .withDataSourceConfiguration(dc)
+       .toTable("MY_TABLE")
+       .withStagingBucketName("BUCKET NAME")
+       .withStorageIntegrationName("STORAGE INTEGRATION NAME")
+       .withUserDataMapper(mapper)
+       .withCreateDisposition(CREATE_NEVER)
+)
+{{< /highlight >}}
+
+#### Table schema disposition
+When the `.withCreateDisposition()` .option is set to `CREATE_IF_NEEDED`, the `.withTableSchema()` option enables specifying the schema for the created target table. 
+A table schema is a list of `SFColumn` objects with name and type corresponding to column type for each column in the table. 
+
+Usage:
+{{< highlight java >}}
+SFTableSchema tableSchema =
+    new SFTableSchema(
+        SFColumn.of("my_date", new SFDate(), true),
+        new SFColumn("id", new SFNumber()),
+        SFColumn.of("name", new SFText(), true));
+
+data.apply(
+   SnowflakeIO.<~>write()
+       .withDataSourceConfiguration(dc)
+       .toTable("MY_TABLE")
+       .withStagingBucketName("BUCKET NAME")
+       .withStorageIntegrationName("STORAGE INTEGRATION NAME")
+       .withUserDataMapper(mapper)
+       .withTableSchema(tableSchema)
+)
+{{< /highlight >}}
+## Reading from Snowflake
+One of the functions of SnowflakeIO is reading Snowflake tables - either full tables via table name or custom data via query. Output of the read transform is a [PCollection](https://beam.apache.org/releases/javadoc/2.17.0/org/apache/beam/sdk/values/PCollection.html) of user-defined data type.
+
+### General usage
+
+The basic `.read()` operation usage:
+{{< highlight java >}}
+PCollection<USER_DATA_TYPE> items = pipeline.apply(
+   SnowflakeIO.<USER_DATA_TYPE>read()
+       .withDataSourceConfiguration(dc)
+       .fromTable("MY_TABLE") // or .fromQuery("QUERY")
+       .withStagingBucketName("BUCKET NAME")
+       .withStorageIntegrationName("STORAGE INTEGRATION NAME")
+       .withCsvMapper(mapper)
+       .withCoder(coder));
+)
+{{< /highlight >}}
+Where all below parameters are required:
+
+- `.withDataSourceConfiguration(...)`
+  - Accepts a DataSourceConfiguration object.
+
+- `.fromTable(...) or .fromQuery(...)`
+  - Specifies a Snowflake table name or custom SQL query.
+
+- `.withStagingBucketName()`
+  - Accepts a cloud bucket name.
+
+-  `.withStorageIntegrationName()`
+  - Accepts a name of a Snowflake storage integration object created according to Snowflake documentation. Example:
+{{< highlight >}}
+CREATE OR REPLACE STORAGE INTEGRATION test_integration 
+TYPE = EXTERNAL_STAGE 
+STORAGE_PROVIDER = GCS 
+ENABLED = TRUE 
+STORAGE_ALLOWED_LOCATIONS = ('gcs://bucket/');
+{{< /highlight >}}
+Then:
+{{< highlight >}}
+.withStorageIntegrationName(test_integration)
+{{< /highlight >}}
+
+- `.withCsvMapper(mapper)`
+  - Accepts a [CSVMapper](https://beam.apache.org/documentation/io/built-in/snowflake/#csvmapper) instance for mapping String[] to USER_DATA_TYPE.
+- `withCoder(coder)`
+  - Accepts the [Coder](https://beam.apache.org/releases/javadoc/2.0.0/org/apache/beam/sdk/coders/Coder.html) for USER_DATA_TYPE.
+
+**Note**:
+SnowflakeIO uses COPY statements behind the scenes to read (using [COPY to location](https://docs.snowflake.net/manuals/sql-reference/sql/copy-into-location.html)) files staged in cloud storage.StagingBucketName will be used as a temporary location for storing CSV files. Those temporary directories will be named `sf_copy_csv_DATE_TIME_RANDOMSUFFIX` and they will be removed automatically once Read operation finishes.
+
+### CSVMapper
+SnowflakeIO uses a [COPY INTO <location>](https://docs.snowflake.net/manuals/sql-reference/sql/copy-into-location.html) statement to move data from a Snowflake table to Google Cloud Storage as CSV files. These files are then downloaded via [FileIO](https://beam.apache.org/releases/javadoc/2.3.0/index.html?org/apache/beam/sdk/io/FileIO.html) and processed line by line. Each line is split into an array of Strings using the [OpenCSV](http://opencsv.sourceforge.net/) library. 
+
+The CSVMapper’s job is to give the user the possibility to convert the array of Strings to a user-defined type, ie. GenericRecord for Avro or Parquet files, or custom POJO.
+
+Example implementation of CsvMapper for GenericRecord:
+{{< highlight java >}}
+static SnowflakeIO.CsvMapper<GenericRecord> getCsvMapper() {
+   return (SnowflakeIO.CsvMapper<GenericRecord>)
+           parts -> {
+               return new GenericRecordBuilder(PARQUET_SCHEMA)
+                       .set("ID", Long.valueOf(parts[0]))
+                       .set("NAME", parts[1])
+                       [...]
+                       .build();
+           };
+}
+{{< /highlight >}}
\ No newline at end of file

From 827ea1200586f506ae188da7a8610e4c7fe0e808 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?S=C5=82awomir=20Andrian?= <>
Date: Fri, 10 Jul 2020 17:01:51 +0200
Subject: [PATCH 2/5] Snowflake IO Connector guide - added requested changes

---
 .../en/documentation/io/built-in/snowflake.md | 380 +-----------------
 1 file changed, 6 insertions(+), 374 deletions(-)

diff --git a/website/www/site/content/en/documentation/io/built-in/snowflake.md b/website/www/site/content/en/documentation/io/built-in/snowflake.md
index f8024c044c20..a115e805c7f8 100644
--- a/website/www/site/content/en/documentation/io/built-in/snowflake.md
+++ b/website/www/site/content/en/documentation/io/built-in/snowflake.md
@@ -102,15 +102,10 @@ Where parameters can be:
 - `.withKeyPairAuth(username, privateKey)`
   - Sets key pair authentication using username and [PrivateKey](https://docs.oracle.com/javase/8/docs/api/java/security/PrivateKey.html)
   - Example: `.withKeyPairAuth("USERNAME",` [PrivateKey](https://docs.oracle.com/javase/8/docs/api/java/security/PrivateKey.html)`)`
-- `.withKeyPairPathAuth(username, privateKeyPath, privateKeyPassphrase)`
-  - Sets key pair authentication using username, path to private key file and passphrase.
-  - Example: `.withKeyPairPathAuth("USERNAME", "PATH/TO/KEY.P8", "PASSPHRASE")`
-- `.withKeyPairRawAuth(username, rawPrivateKey, privateKeyPassphrase)`
-  - Sets key pair authentication using username, private key and passphrase.
-  - Example: `.withKeyPairRawAuth("USERNAME", "PRIVATE_KEY", "PASSPHRASE")`
 
 
 **Note** - either `.withUrl(...)` or `.withServerName(...)` **is required**.
+
 ## Pipeline options
 Use Beam’s [Pipeline options](https://beam.apache.org/releases/javadoc/2.17.0/org/apache/beam/sdk/options/PipelineOptions.html) to set options via the command line.
 ### Snowflake Pipeline options
@@ -126,10 +121,6 @@ Snowflake IO library supports following options that can be passed via the [comm
 
 `--password` Required for username/password authentication only.
 
-`--privateKeyPath` Path to Private Key file. Required for Private Key authentication only.
-
-`--rawPrivateKey` Private Key. Required for Private Key authentication only.
-
 `--privateKeyPassphrase` Private Key's passphrase. Required for Private Key authentication only.
 
 `--stagingBucketName` External bucket path ending with `/`. I.e. `gs://bucket/`. Sub-directories are allowed.
@@ -152,305 +143,6 @@ Snowflake IO library supports following options that can be passed via the [comm
 
 `--portNumber` Port number. Optional.
 
-`--loginTimeout` Login timeout. Optional.
-
-`--snowPipe` SnowPipe name. Optional.
-
-In the connector code SnowflakePipelineOptions looks following (and it can be [extended](https://beam.apache.org/documentation/io/built-in/snowflake/#extending-pipeline-options)):
-{{< highlight java >}}
-public interface SnowflakePipelineOptions extends PipelineOptions, StreamingOptions {
-  String BASIC_CONNECTION_INFO_VALIDATION_GROUP = "BASIC_CONNECTION_INFO_GROUP";
-  String AUTH_VALIDATION_GROUP = "AUTH_VALIDATION_GROUP";
-
-  @Description(
-      "Snowflake's JDBC-like url including account name and region without any parameters.")
-  @Validation.Required(groups = BASIC_CONNECTION_INFO_VALIDATION_GROUP)
-  String getUrl();
-
-  void setUrl(String url);
-
-  @Description("Server Name - full server name with account, zone and domain.")
-  @Validation.Required(groups = BASIC_CONNECTION_INFO_VALIDATION_GROUP)
-  ValueProvider<String> getServerName();
-
-  void setServerName(ValueProvider<String> serverName);
-
-  @Description("Username. Required for username/password and Private Key authentication.")
-  @Validation.Required(groups = AUTH_VALIDATION_GROUP)
-  ValueProvider<String> getUsername();
-
-  void setUsername(ValueProvider<String> username);
-
-  @Description("OAuth token. Required for OAuth authentication only.")
-  @Validation.Required(groups = AUTH_VALIDATION_GROUP)
-  String getOauthToken();
-
-  void setOauthToken(String oauthToken);
-
-  @Description("Password. Required for username/password authentication only.")
-  @Default.String("")
-  ValueProvider<String> getPassword();
-
-  void setPassword(ValueProvider<String> password);
-
-  @Description("Path to Private Key file. Required for Private Key authentication only.")
-  @Default.String("")
-  String getPrivateKeyPath();
-
-  void setPrivateKeyPath(String privateKeyPath);
-
-  @Description("Private key. Required for Private Key authentication only.")
-  @Default.String("")
-  ValueProvider<String> getRawPrivateKey();
-
-  void setRawPrivateKey(ValueProvider<String> rawPrivateKey);
-
-  @Description("Private Key's passphrase. Required for Private Key authentication only.")
-  @Default.String("")
-  ValueProvider<String> getPrivateKeyPassphrase();
-
-  void setPrivateKeyPassphrase(ValueProvider<String> keyPassphrase);
-
-  @Description("Warehouse to use. Optional.")
-  @Default.String("")
-  ValueProvider<String> getWarehouse();
-
-  void setWarehouse(ValueProvider<String> warehouse);
-
-  @Description("Database name to connect to. Optional.")
-  @Default.String("")
-  ValueProvider<String> getDatabase();
-
-  void setDatabase(ValueProvider<String> database);
-
-  @Description("Schema to use. Optional.")
-  @Default.String("")
-  ValueProvider<String> getSchema();
-
-  void setSchema(ValueProvider<String> schema);
-
-  @Description("Table to use. Optional.")
-  @Default.String("")
-  ValueProvider<String> getTable();
-
-  void setTable(ValueProvider<String> table);
-
-  @Description("Query to use. Optional.")
-  @Default.String("")
-  ValueProvider<String> getQuery();
-
-  void setQuery(ValueProvider<String> query);
-
-  @Description("Role to use. Optional.")
-  @Default.String("")
-  ValueProvider<String> getRole();
-
-  void setRole(ValueProvider<String> role);
-
-  @Description("Authenticator to use. Optional.")
-  @Default.String("")
-  String getAuthenticator();
-
-  void setAuthenticator(String authenticator);
-
-  @Description("Port number. Optional.")
-  @Default.String("")
-  String getPortNumber();
-
-  void setPortNumber(String portNumber);
-
-  @Description("Login timeout. Optional.")
-  @Default.String("")
-  String getLoginTimeout();
-
-  void setLoginTimeout(String loginTimeout);
-
-  @Description("Temporary GCS bucket name.")
-  ValueProvider<String> getStagingBucketName();
-
-  void setStagingBucketName(ValueProvider<String> stagingBucketName);
-
-  @Description("Storage integration name")
-  ValueProvider<String> getStorageIntegrationName();
-
-  void setStorageIntegrationName(ValueProvider<String> storageIntegrationName);
-
-  @Description("SnowPipe name. Optional.")
-  ValueProvider<String> getSnowPipe();
-
-  void setSnowPipe(ValueProvider<String> snowPipe);
-}
-{{< /highlight >}}
-### Using Pipeline options
-To use Pipeline options, you must configure them as follows:
-
-{{< highlight java>}}
-SnowflakePipelineOptions options = PipelineOptionsFactory
-    .fromArgs(args)
-    .withValidation()
-    .as(SnowflakePipelineOptions.class);
-{{< /highlight >}}
-
-All the below parameters are required:
-- `.fromArgs()`
-  - GNU style command line arguments, 
-  - Example: `--project=myproject --database=test`
-- `.withValidation()`
-  - which validates that PipelineOptions confirms all criteria from the passed in interface
-- `.as()`
-  - a class of used pipeline options
-  - Example: `.as(SnowflakePipelineOptions.class)`
-
-Then create your pipeline using created options:
-
-{{< highlight java >}}
-Pipeline pipeline = Pipeline.create(options);
-{{< /highlight >}}
-
-Example of accessing pipeline options in code:
-
-{{< highlight java >}}
-String stagingBucketName = options.getStagingBucketName();
-{{< /highlight >}}
-
-Pipelines in tests:
-{{< highlight java >}}
-PipelineOptionsFactory.register(SnowflakePipelineOptions.class);
-    options = TestPipeline
-                .testingPipelineOptions()                                  
-                .as(SnowflakePipelineOptions.class);
-{{< /highlight >}}
-
-Then use them in pipeline run:
-{{< highlight java >}}
-PipelineResult pipelineResult = pipeline.run(options);
-{{< /highlight >}}
-
-### Extending pipeline options
-Extend the SnowflakePipelineOptions with your own custom options to access additional parameters in your code.
-
-Example of extending the Pipeline options:
-{{< highlight java >}}
-public interface BatchTestPipelineOptions extends SnowflakePipelineOptions {
-  @Description("Table name to connect to.")
-  String getTable();
-
-  void setTable(String table);
-}
-{{< /highlight >}}
-
-The following example shows how to insert the custom options:
-{{< highlight java >}}
-ExampleSnowflakePipelineOptions options = PipelineOptionsFactory
-.fromArgs(args)
-.withValidation()
-.as(ExampleSnowflakePipelineOptions.class);
-
-options.getTable();
-{{< /highlight >}}
-
-And in case of testing:
-{{< highlight java >}}
-PipelineOptionsFactory.register(ExampleSnowflakePipelineOptions.class);
-    options = TestPipeline
-    .testingPipelineOptions()
-                .as(ExampleSnowflakePipelineOptions.class);
-{{< /highlight >}}
-
-### Running main command with Pipeline options
-To pass Pipeline options via the command line, use `--args` in a gradle command as follows:
-
-{{< highlight >}}
-./gradle run 
-    --args="
-        --serverName=<SNOWFLAKE SERVER NAME>  
-           Example: --serverName=account.region.gcp.snowflakecomputing.com
-        --username=<SNOWFLAKE USERNAME> 
-           Example: --username=testuser
-        --password=<SNOWFLAKE PASSWORD> 
-           Example: --password=mypassword
-        --database=<SNOWFLAKE DATABASE> 
-           Example: --database=TEST_DATABASE
-        --schema=<SNOWFLAKE SCHEMA> 
-           Example: --schema=public
-        --table=<SNOWFLAKE TABLE IN DATABASE> 
-           Example: --table=TEST_TABLE
-        --query=<IF NOT TABLE THEN QUERY> 
-           Example: --query=‘SELECT column FROM TABLE’
-        --storageIntegrationName=<SNOWFLAKE STORAGE INTEGRATION NAME> 
-           Example: --storageIntegrationName=my_integration
-        --stagingBucketName=<GCS BUCKET NAME>
-           Example: --stagingBucketName=gs://my_gcp_bucket/
-        --runner=<DirectRunner/DataflowRunner>
-           Example: --runner=DataflowRunner
-        --project=<FOR DATAFLOW RUNNER: GCP PROJECT NAME> 
-           Example: --project=my_project
-        --tempLocation=<FOR DATAFLOW RUNNER: GCS TEMP LOCATION STARTING
-                        WITH gs://…>
-           Example: --tempLocation=gs://my_bucket/temp/
-        --region=<FOR DATAFLOW RUNNER: GCP REGION> 
-           Example: --region=us-east-1
-        --appName=<OPTIONAL: DATAFLOW JOB NAME PREFIX>
-           Example: --appName=my_job
-    "
-    {{< /highlight >}}
-Then in the code it is possible to access the parameters with arguments using the options.getStagingBucketName(); command.
-
-### Running test command with Pipeline options
-To pass Pipeline options via the command line, use `-DintegrationTestPipelineOptions` in a gradle command as follows:
-{{< highlight >}}
-./gradlew test --tests nameOfTest 
--DintegrationTestPipelineOptions='[
-  "--serverName=<SNOWFLAKE SERVER NAME>",
-      Example: --serverName=account.region.gcp.snowflakecomputing.com
-  "--username=<SNOWFLAKE USERNAME>",
-      Example: --username=testuser
-  "--password=<SNOWFLAKE PASSWORD>",
-      Example: --password=mypassword
-  "--schema=<SNOWFLAKE SCHEMA>",
-      Example: --schema=PUBLIC
-  "--table=<SNOWFLAKE TABLE IN DATABASE>",
-      Example: --table=TEST_TABLE
-  "--database=<SNOWFLAKE DATABASE>",
-      Example: --database=TEST_DATABASE
-  "--storageIntegrationName=<SNOWFLAKE STORAGE INTEGRATION NAME>", 
-      Example: --storageIntegrationName=my_integration
-  "--stagingBucketName=<GCS BUCKET NAME>",
-      Example: --stagingBucketName=gs://my_gcp_bucket
-  "--externalLocation=<GCS BUCKET URL STARTING WITH GS://>",
-      Example: --tempLocation=gs://my_bucket/temp/
-]' --no-build-cache
-{{< /highlight >}}
-
-Where all parameters are starting with “--”, they are surrounded with double quotation and separated with comma:
-
-- `--serverName=<SNOWFLAKE SERVER NAME>`
-  - Specifies the full name of your account (provided by Snowflake). Note that your full account name might include additional segments that identify the region and cloud platform where your account is hosted. 
-  - Example: `--serverName=xy12345.eu-west-1.gcp..snowflakecomputing.com`
-
-- `--username=<SNOWFLAKE USERNAME>`
-  - Specifies the login name of the user.
-  - Example: `--username=my_username`
-
-- `--password=<SNOWFLAKE PASSWORD>`
-  - Specifies the password for the specified user.
-  - Example: `--password=my_secret`
-
-- `--schema=<SNOWFLAKE SCHEMA>`
-  - Specifies the schema to use for the specified database once connected. The specified schema should be an existing schema for which the specified user’s role has privileges.
-  - Example: `--schema=PUBLIC`
-
-- `--table=<SNOWFLAKE TABLE IN DATABASE>`
-  - Example: `--table=MY_TABLE`
-
-- `--database=<SNOWFLAKE DATABASE>`
-  - Specifies the database to use once connected. The specified database should be an existing database for which the specified user’s role has privileges.
-  - Example: `--database=MY_DATABASE`
-
-- `--storageIntegrationName=<SNOWFLAKE STORAGE INTEGRATION NAME>`
-  - Name of storage integration created in [Snowflake](https://docs.snowflake.com/en/sql-reference/sql/create-storage-integration.html) for a cloud storage of choice.
-  - Example: `--storageIntegrationName=my_google_integration`
-
 ## Running pipelines on Dataflow
 By default, pipelines are run on [Direct Runner](https://beam.apache.org/documentation/runners/direct/) on your local machine. To run a pipeline on [Google Dataflow](https://cloud.google.com/dataflow/), you must provide the following Pipeline options:
 
@@ -475,60 +167,6 @@ More pipeline options for Dataflow can be found [here](https://beam.apache.org/r
 
 **Important**: Please acknowledge [Google Dataflow pricing](Important: Please acknowledge Google Dataflow pricing).
 
-###Running pipeline templates on Dataflow
-Google Dataflow is supporting [template](https://cloud.google.com/dataflow/docs/guides/templates/overview) creation which means staging pipelines on Cloud Storage and running them with ability to pass runtime parameters that are only available during pipeline execution.
-
-The process of creating own Dataflow template is following
-1. Create your own pipeline. 
-2. Create [Dataflow template](https://cloud.google.com/dataflow/docs/guides/templates/creating-templates#creating-and-staging-templates) with checking which options SnowflakeIO is supporting at runtime.
-3. Run a Dataflow template using [Cloud Console](https://cloud.google.com/dataflow/docs/guides/templates/running-templates#using-the-cloud-console), [REST API](https://cloud.google.com/dataflow/docs/guides/templates/running-templates#using-the-rest-api) or [gcloud](https://cloud.google.com/dataflow/docs/guides/templates/running-templates#using-gcloud).
-
-Currently, SnowflakeIO supports following options at runtime:
-
-- `--serverName` Full server name with account, zone and domain.
-
-- `--username` Required for username/password and Private Key authentication.
-
-- `--password` Required for username/password authentication only.
-
-- `--rawPrivateKey` Private Key file. Required for Private Key authentication only.
-
-- `--privateKeyPassphrase` Private Key's passphrase. Required for Private Key authentication only.
-
-- `--stagingBucketName` external bucket path ending with `/`. I.e. `gs://bucket/`. Sub-directories are allowed.
-
-- `--storageIntegrationName` Storage integration name.
-
-- `--warehouse` Warehouse to use. Optional.
-
-- `--database` Database name to connect to. Optional.
-
-- `--schema` Schema to use. Optional.
-
-- `--table` Table to use. Optional. Note: table is not in default pipeline options.
-
-- `--query` Query to use. Optional. Note: query is not in default pipeline options.
-
-- `--role` Role to use. Optional.
-
-- `--snowPipe` SnowPipe name. Optional.
-
-**Note**: table and query is not in pipeline options by default, it may be added by [extending](https://beam.apache.org/documentation/io/built-in/snowflake/#extending-pipeline-options) PipelineOptions.
-
-Currently, SnowflakeIO **doesn't support** following options at runtime:
-
-- `--url` Snowflake's JDBC-like url including account name and region without any parameters.
-
-- `--oauthToken` Required for OAuth authentication only.
-
-- `--privateKeyPath` Path to Private Key file. Required for Private Key authentication only.
-
-- `--authenticator` Authenticator to use. Optional.
-
-- `--portNumber` Port number. Optional.
-
-- `--loginTimeout` Login timeout. Optional.
-
 ## Writing to Snowflake tables
 One of the functions of SnowflakeIO is writing to Snowflake tables. This transformation enables you to finish the Beam pipeline with an output operation that sends the user's [PCollection](https://beam.apache.org/releases/javadoc/2.17.0/org/apache/beam/sdk/values/PCollection.html) to your Snowflake database.
 ### Batch write (from a bounded source)
@@ -537,7 +175,7 @@ The basic .`write()` operation usage is as follows:
 data.apply(
    SnowflakeIO.<type>write()
        .withDataSourceConfiguration(dc)
-       .toTable("MY_TABLE")
+       .to("MY_TABLE")
        .withStagingBucketName("BUCKET NAME")
        .withStorageIntegrationName("STORAGE INTEGRATION NAME")
        .withUserDataMapper(mapper)
@@ -572,12 +210,6 @@ Then:
 **Note**:
 SnowflakeIO uses COPY statements behind the scenes to write (using [COPY to table](https://docs.snowflake.net/manuals/sql-reference/sql/copy-into-table.html)). StagingBucketName will be used to save CSV files which will end up in Snowflake. Those CSV files will be saved under the “stagingBucketName” path.
 
-**Optional** for batching:
-- `.withQuotationMark()` 
-  - Default value: ‘ (single quotation mark).
-  - Accepts String with one character. It will surround all text (String) fields saved to CSV. It should be one of the accepted characters by [Snowflake’s](https://docs.snowflake.com/en/sql-reference/sql/create-file-format.html) [FIELD_OPTIONALLY_ENCLOSED_BY](https://docs.snowflake.com/en/sql-reference/sql/create-file-format.html) parameter (double quotation mark, single quotation mark or none).
-  - Example: `.withQuotationMark("'")`
-
 ### UserDataMapper function
 The UserDataMapper function is required to map data from a PCollection to an array of String values before the `write()` operation saves the data to temporary .csv files. For example:
 {{< highlight java >}}
@@ -596,7 +228,7 @@ String query = "SELECT t.$1 from YOUR_TABLE;";
 data.apply(
    SnowflakeIO.<~>write()
        .withDataSourceConfiguration(dc)
-       .toTable("MY_TABLE")
+       .to("MY_TABLE")
        .withStagingBucketName("BUCKET NAME")
        .withStorageIntegrationName("STORAGE INTEGRATION NAME")
        .withUserDataMapper(mapper)
@@ -618,7 +250,7 @@ Example of usage:
 data.apply(
    SnowflakeIO.<~>write()
        .withDataSourceConfiguration(dc)
-       .toTable("MY_TABLE")
+       .to("MY_TABLE")
        .withStagingBucketName("BUCKET NAME")
        .withStorageIntegrationName("STORAGE INTEGRATION NAME")
        .withUserDataMapper(mapper)
@@ -638,7 +270,7 @@ Usage:
 data.apply(
    SnowflakeIO.<~>write()
        .withDataSourceConfiguration(dc)
-       .toTable("MY_TABLE")
+       .to("MY_TABLE")
        .withStagingBucketName("BUCKET NAME")
        .withStorageIntegrationName("STORAGE INTEGRATION NAME")
        .withUserDataMapper(mapper)
@@ -661,7 +293,7 @@ SFTableSchema tableSchema =
 data.apply(
    SnowflakeIO.<~>write()
        .withDataSourceConfiguration(dc)
-       .toTable("MY_TABLE")
+       .to("MY_TABLE")
        .withStagingBucketName("BUCKET NAME")
        .withStorageIntegrationName("STORAGE INTEGRATION NAME")
        .withUserDataMapper(mapper)

From edcb2ebb539810b032eaa29d0f4fe018e244fb29 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?S=C5=82awomir=20Andrian?= <>
Date: Thu, 16 Jul 2020 12:45:59 +0200
Subject: [PATCH 3/5] Adjusted documentation to the version that is available
 in Apache Beam

---
 .../en/documentation/io/built-in/snowflake.md | 24 +++++++------------
 1 file changed, 9 insertions(+), 15 deletions(-)

diff --git a/website/www/site/content/en/documentation/io/built-in/snowflake.md b/website/www/site/content/en/documentation/io/built-in/snowflake.md
index a115e805c7f8..078a373f4d91 100644
--- a/website/www/site/content/en/documentation/io/built-in/snowflake.md
+++ b/website/www/site/content/en/documentation/io/built-in/snowflake.md
@@ -46,12 +46,11 @@ To use username/password authentication in SnowflakeIO, invoke your pipeline wit
 --username=<USERNAME> --password=<PASSWORD>
 {{< /highlight >}}
 ### Key pair
-To use this authentication method, you must first generate a key pair and associate the public key with the Snowflake user that will connect using the IO transform. For instructions,  see the [Snowflake documentation](https://docs.snowflake.com/en/user-guide/jdbc-configure.html).
+To use this authentication method, you must first generate a key pair and associate the public key with the Snowflake user that will connect using the IO transform. For instructions, see the [Snowflake documentation](https://docs.snowflake.com/en/user-guide/jdbc-configure.html).
 
-To use key pair authentication with SnowflakeIO, invoke your pipeline with one of the following set of Pipeline options:
+To use key pair authentication with SnowflakeIO, invoke your pipeline with following Pipeline options:
 {{< highlight >}}
 --username=<USERNAME> --privateKeyPath=<PATH_TO_P8_FILE> --privateKeyPassphrase=<PASSWORD_FOR_KEY>
---username=<USERNAME> --rawPrivateKey=<PRIVATE_KEY> --privateKeyPassphrase=<PASSWORD_FOR_KEY>
 {{< /highlight >}}
 
 ### OAuth token
@@ -69,7 +68,7 @@ DataSource configuration is required in both read and write object for configuri
 Create the DataSource configuration:
 {{< highlight java >}}
  SnowflakeIO.DataSourceConfiguration
-            .create()
+            .create(SnowflakeCredentialsFactory.of(options))
             .withUrl(options.getUrl())
             .withServerName(options.getServerName())
             .withDatabase(options.getDatabase())
@@ -93,15 +92,6 @@ Where parameters can be:
 - `.withSchema(...)`
   - Name of the schema in the database to use. This parameter is optional.
   - Example: `.withSchema("PUBLIC")`
-- `.withUsernamePasswordAuth(username, password)`
-  - Sets username/password authentication.
-  - Example: `.withUsernamePasswordAuth("USERNAME", "PASSWORD")`
-- `.withOAuth(token)`
-  - Sets OAuth authentication.
-  - Example: `.withOAuth("TOKEN")`
-- `.withKeyPairAuth(username, privateKey)`
-  - Sets key pair authentication using username and [PrivateKey](https://docs.oracle.com/javase/8/docs/api/java/security/PrivateKey.html)
-  - Example: `.withKeyPairAuth("USERNAME",` [PrivateKey](https://docs.oracle.com/javase/8/docs/api/java/security/PrivateKey.html)`)`
 
 
 **Note** - either `.withUrl(...)` or `.withServerName(...)` **is required**.
@@ -121,6 +111,8 @@ Snowflake IO library supports following options that can be passed via the [comm
 
 `--password` Required for username/password authentication only.
 
+`--privateKeyPath` Path to Private Key file. Required for Private Key authentication only.
+
 `--privateKeyPassphrase` Private Key's passphrase. Required for Private Key authentication only.
 
 `--stagingBucketName` External bucket path ending with `/`. I.e. `gs://bucket/`. Sub-directories are allowed.
@@ -143,6 +135,8 @@ Snowflake IO library supports following options that can be passed via the [comm
 
 `--portNumber` Port number. Optional.
 
+`--loginTimeout` Login timeout. Optional.
+
 ## Running pipelines on Dataflow
 By default, pipelines are run on [Direct Runner](https://beam.apache.org/documentation/runners/direct/) on your local machine. To run a pipeline on [Google Dataflow](https://cloud.google.com/dataflow/), you must provide the following Pipeline options:
 
@@ -187,7 +181,7 @@ All the below parameters are required:
 
 - `.withDataSourceConfiguration()` Accepts a DatasourceConfiguration object.
 
-- `.toTable()` Accepts the target Snowflake table name.
+- `.to()` Accepts the target Snowflake table name.
 
 - `.withStagingBucketName()` Accepts a cloud bucket path ended with slash.
  -Example: `.withStagingBucketName("gs://mybucket/my/dir/")`
@@ -344,7 +338,7 @@ Then:
 
 - `.withCsvMapper(mapper)`
   - Accepts a [CSVMapper](https://beam.apache.org/documentation/io/built-in/snowflake/#csvmapper) instance for mapping String[] to USER_DATA_TYPE.
-- `withCoder(coder)`
+- `.withCoder(coder)`
   - Accepts the [Coder](https://beam.apache.org/releases/javadoc/2.0.0/org/apache/beam/sdk/coders/Coder.html) for USER_DATA_TYPE.
 
 **Note**:

From e21ff73d12a7c0ef09e8c7a5195ba09e697825d8 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?S=C5=82awomir=20Andrian?= <>
Date: Tue, 4 Aug 2020 17:24:48 +0200
Subject: [PATCH 4/5] Added link to guide for SnowflakeIO documentation

---
 website/www/site/data/io_matrix.yaml | 1 +
 1 file changed, 1 insertion(+)

diff --git a/website/www/site/data/io_matrix.yaml b/website/www/site/data/io_matrix.yaml
index 5923a86ab664..84637c615850 100644
--- a/website/www/site/data/io_matrix.yaml
+++ b/website/www/site/data/io_matrix.yaml
@@ -305,6 +305,7 @@ categories:
             url: https://beam.apache.org/releases/pydoc/current/apache_beam.io.gcp.datastore.v1new.datastoreio.html
       - transform: SnowflakeIO
         description: Experimental Transforms for reading from and writing to [Snowflake](https://www.snowflake.com/).
+        docs: /documentation/io/built-in/snowflake
         implementations:
           - language: java
             name: org.apache.beam.sdk.io.snowflake.SnowflakeIO

From dac8b38c51e3ef8c076690feabc580b207f33492 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?S=C5=82awomir=20Andrian?= <>
Date: Thu, 6 Aug 2020 13:12:02 +0200
Subject: [PATCH 5/5] Added link in the table of contents for Snowflake I/O
 connector

---
 .../www/site/layouts/partials/section-menu/en/documentation.html | 1 +
 1 file changed, 1 insertion(+)

diff --git a/website/www/site/layouts/partials/section-menu/en/documentation.html b/website/www/site/layouts/partials/section-menu/en/documentation.html
index c8d0da9c6abb..fd35a21bd879 100644
--- a/website/www/site/layouts/partials/section-menu/en/documentation.html
+++ b/website/www/site/layouts/partials/section-menu/en/documentation.html
@@ -74,6 +74,7 @@
             <li><a href="/documentation/io/built-in/hadoop/">Hadoop Input/Output Format IO</a></li>
             <li><a href="/documentation/io/built-in/hcatalog/">HCatalog IO</a></li>
             <li><a href="/documentation/io/built-in/google-bigquery/">Google BigQuery I/O connector</a></li>
+            <li><a href="/documentation/io/built-in/snowflake/">Snowflake I/O connector</a></li>
           </ul>
        </li>
 
