You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2020/07/13 04:30:08 UTC
[GitHub] [beam] chamikaramj commented on a change in pull request #12149: [BEAM-9897] Add cross-language support to SnowflakeIO.Read
chamikaramj commented on a change in pull request #12149:
URL: https://github.com/apache/beam/pull/12149#discussion_r453425805
##########
File path: sdks/java/container/build.gradle
##########
@@ -49,6 +49,7 @@ dependencies {
dockerDependency project(":sdks:java:io:kafka")
// This dependency is set to 'provided' scope in :sdks:java:io:kafka
dockerDependency library.java.kafka_clients
+ dockerDependency project(path: ":sdks:java:io:snowflake", configuration: "shadow")
Review comment:
Ditto.
##########
File path: sdks/java/container/build.gradle
##########
@@ -60,6 +61,8 @@ task copyDockerfileDependencies(type: Copy) {
rename 'beam-sdks-java-harness-.*.jar', 'beam-sdks-java-harness.jar'
rename 'beam-sdks-java-io-kafka.*.jar', 'beam-sdks-java-io-kafka.jar'
rename 'kafka-clients.*.jar', 'kafka-clients.jar'
+ rename 'beam-sdks-java-io-snowflake.*SNAPSHOT.jar', 'beam-sdks-java-io-snowflake.jar'
Review comment:
Ditto.
##########
File path: sdks/java/io/snowflake/src/main/java/org/apache/beam/sdk/io/snowflake/credentials/OAuthTokenSnowflakeCredentials.java
##########
@@ -28,4 +30,9 @@ public OAuthTokenSnowflakeCredentials(String token) {
public String getToken() {
return token;
}
+
+ @Override
Review comment:
I suggest moving changes to credentials to a separate PR since this seems to be unrelated to x-lang changes.
##########
File path: sdks/java/container/boot.go
##########
@@ -122,6 +122,7 @@ func main() {
filepath.Join(jarsDir, "beam-sdks-java-harness.jar"),
filepath.Join(jarsDir, "beam-sdks-java-io-kafka.jar"),
filepath.Join(jarsDir, "kafka-clients.jar"),
+ filepath.Join(jarsDir, "beam-sdks-java-io-snowflake.jar"),
Review comment:
You should not need this. You just have to make sure that dependencies correctly get determined (and staged) during cross language expansion.
##########
File path: sdks/java/expansion-service/build.gradle
##########
@@ -42,6 +42,8 @@ dependencies {
compile library.java.slf4j_api
runtimeOnly library.java.slf4j_jdk14
testCompile library.java.junit
+
+ runtime project(":sdks:java:io:snowflake")
Review comment:
Why are we adding a dependency here ?
##########
File path: sdks/java/io/snowflake/src/main/java/org/apache/beam/sdk/io/snowflake/SnowflakeIO.java
##########
@@ -960,26 +953,44 @@ public static DataSourceConfiguration create(DataSource dataSource) {
* @param credentials - an instance of {@link SnowflakeCredentials}.
*/
public static DataSourceConfiguration create(SnowflakeCredentials credentials) {
- if (credentials instanceof UsernamePasswordSnowflakeCredentials) {
- return new AutoValue_SnowflakeIO_DataSourceConfiguration.Builder()
- .setValidate(true)
- .setUsername(((UsernamePasswordSnowflakeCredentials) credentials).getUsername())
- .setPassword(((UsernamePasswordSnowflakeCredentials) credentials).getPassword())
- .build();
- } else if (credentials instanceof OAuthTokenSnowflakeCredentials) {
- return new AutoValue_SnowflakeIO_DataSourceConfiguration.Builder()
- .setValidate(true)
- .setOauthToken(((OAuthTokenSnowflakeCredentials) credentials).getToken())
- .build();
- } else if (credentials instanceof KeyPairSnowflakeCredentials) {
- return new AutoValue_SnowflakeIO_DataSourceConfiguration.Builder()
- .setValidate(true)
- .setUsername(((KeyPairSnowflakeCredentials) credentials).getUsername())
- .setPrivateKey(((KeyPairSnowflakeCredentials) credentials).getPrivateKey())
- .build();
- }
- throw new IllegalArgumentException(
- "Can't create DataSourceConfiguration from given credentials");
+ return credentials.createSnowflakeDataSourceConfiguration();
+ }
+
+ /**
+ * Creates {@link DataSourceConfiguration} from instance of {@link
Review comment:
Why are we implementing separate methods here ? Also probably this should be a separate PR.
##########
File path: sdks/java/io/snowflake/src/main/java/org/apache/beam/sdk/io/snowflake/xlang/Configuration.java
##########
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.beam.sdk.io.snowflake.xlang;
Review comment:
Probably just leave this in the package "org.apache.beam.sdk.io.snowflake" ? Is there a need to add a new package ? (and if so probably use long form "crosslanguage").
##########
File path: sdks/java/io/snowflake/src/main/java/org/apache/beam/sdk/io/snowflake/xlang/ExternalRead.java
##########
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.beam.sdk.io.snowflake.xlang;
+
+import com.google.auto.service.AutoService;
+import java.io.Serializable;
+import java.nio.charset.Charset;
+import java.util.Map;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.expansion.ExternalTransformRegistrar;
+import org.apache.beam.sdk.io.snowflake.SnowflakeIO;
+import org.apache.beam.sdk.io.snowflake.credentials.SnowflakeCredentials;
+import org.apache.beam.sdk.io.snowflake.credentials.SnowflakeCredentialsFactory;
+import org.apache.beam.sdk.transforms.ExternalTransformBuilder;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.SerializableFunction;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/** Exposes {@link SnowflakeIO.Read} as an external transform for cross-language usage. */
+@Experimental
+@AutoService(ExternalTransformRegistrar.class)
+public final class ExternalRead implements ExternalTransformRegistrar {
Review comment:
Probably call this SnowflakeReadRegistrar.
##########
File path: sdks/python/apache_beam/io/external/snowflake.py
##########
@@ -0,0 +1,144 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+#
+
+# pytype: skip-file
+
+from __future__ import absolute_import
+
+import typing
+
+from past.builtins import unicode
+
+import apache_beam as beam
+from apache_beam.transforms.external import BeamJarExpansionService
+from apache_beam.transforms.external import ExternalTransform
+from apache_beam.transforms.external import NamedTupleBasedPayloadBuilder
+
+__all__ = ['ReadFromSnowflake']
+
+
+def default_io_expansion_service():
+ return BeamJarExpansionService('sdks:java:io:expansion-service:shadowJar')
Review comment:
You have to either add snowflake jars to "io:expansion-service" or expose a new Snowflake shadow jar that this wrapper can use (for expansion and for staging for runtime). What will be the size diff of "sdks:java:io:expansion-service:shadowJar" after including Snowflake ? If that is large we should go for the second option since this jar is shared by a bunch of cross-language wrappers.
##########
File path: sdks/python/apache_beam/io/external/snowflake.py
##########
@@ -0,0 +1,144 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+#
+
+# pytype: skip-file
+
+from __future__ import absolute_import
+
+import typing
+
+from past.builtins import unicode
+
+import apache_beam as beam
+from apache_beam.transforms.external import BeamJarExpansionService
+from apache_beam.transforms.external import ExternalTransform
+from apache_beam.transforms.external import NamedTupleBasedPayloadBuilder
+
+__all__ = ['ReadFromSnowflake']
+
+
+def default_io_expansion_service():
+ return BeamJarExpansionService('sdks:java:io:expansion-service:shadowJar')
+
+
+ReadFromSnowflakeSchema = typing.NamedTuple(
+ 'WriteToSnowflakeSchema',
+ [
+ ('server_name', unicode),
+ ('schema', unicode),
+ ('database', unicode),
+ ('staging_bucket_name', unicode),
+ ('storage_integration_name', unicode),
+ ('username', typing.Optional[unicode]),
+ ('password', typing.Optional[unicode]),
+ ('private_key_path', typing.Optional[unicode]),
+ ('private_key_passphrase', typing.Optional[unicode]),
+ ('o_auth_token', typing.Optional[unicode]),
+ ('table', typing.Optional[unicode]),
+ ('query', typing.Optional[unicode]),
+ ])
+
+
+class ReadFromSnowflake(beam.PTransform):
+ """An external PTransform which reads from Snowflake."""
+
+ URN = 'beam:external:java:snowflake:read:v1'
+
+ def __init__(
+ self,
+ server_name,
+ schema,
+ database,
+ staging_bucket_name,
+ storage_integration_name,
+ csv_mapper,
+ username=None,
+ password=None,
+ private_key_path=None,
+ private_key_passphrase=None,
+ o_auth_token=None,
+ table=None,
+ query=None,
+ expansion_service=None):
+ """
+ Initializes a read operation from Snowflake.
+
+ Required parameters:
+ :param server_name: full Snowflake server name with the following format
+ account.region.gcp.snowflakecomputing.com.
+ :param schema: name of the Snowflake schema in the database to use.
+ :param database: name of the Snowflake database to use.
+ :param staging_bucket_name: name of the Google Cloud Storage bucket.
+ Bucket 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.
+ :param storage_integration_name: is the name of storage integration
+ object created according to Snowflake documentation.
+ :param csv_mapper: specifies a function which must translate
+ user-defined object to array of strings.
+ SnowflakeIO uses a COPY INTO <location> statement to
+ move data from a Snowflake table to Google Cloud Storage as CSV files.
+ These files are then downloaded via FileIO and processed line by line.
+ Each line is split into an array of Strings using the OpenCSV
+ The csv_mapper function 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 objects.
+ Example:
+ ```
+ def csv_mapper(strings_array):
+ return User(strings_array[0], int(strings_array[1])))
+ ```
+ :param table or query: specifies a Snowflake table name or custom SQL query
+ :param expansion_service: specifies URL of expansion service.
+
+ Authentication parameters:
+ It's required to pass one of the following combinations of valid parameters:
+ :param username and password: specifies username and password
+ for username/password authentication method.
+ :param private_key_path and private_key_passphrase:
+ specifies a private key file and password
+ for key/ pair authentication method.
+ :param o_auth_token: specifies access token for OAuth authentication method.
+ """
+
+ self.params = ReadFromSnowflakeSchema(
Review comment:
Have you tested this against portable Flink/Spark and or Dataflow. Please mention here the runners this have been tested for and supported. Also mention details about prerequisites for the user. See following for an example.
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/io/kafka.py#L18
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org