Samrat002 commented on code in PR #206: URL: https://github.com/apache/flink-connector-aws/pull/206#discussion_r2253162758
########## docs/content/docs/connectors/table/glue.md: ########## @@ -0,0 +1,433 @@ +--- +title: "AWS Glue Catalog" +weight: 11 +type: docs +aliases: + - /dev/table/connectors/glue.html +--- +<!-- +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. +--> + +# AWS Glue Catalog + +The AWS Glue Catalog provides a way to use [AWS Glue](https://aws.amazon.com/glue) as a catalog for Apache Flink. +This allows users to access Glue's metadata store directly from Flink SQL and Table API. + +## Features + +- Register AWS Glue as a catalog in Flink applications +- Access Glue databases and tables through Flink SQL +- Support for various AWS data sources (S3, Kinesis, MSK) +- Mapping between Flink and AWS Glue data types +- Compatibility with Flink's Table API and SQL interface + +## Dependencies + +{{< sql_download_table "glue" >}} + +## Prerequisites + +Before getting started, ensure you have the following: + +- **AWS account** with appropriate permissions for AWS Glue and other required services +- **AWS credentials** properly configured Review Comment: Can you add a simlink to `How to configure from AWS documentation ?` ########## docs/content/docs/connectors/table/glue.md: ########## @@ -0,0 +1,433 @@ +--- +title: "AWS Glue Catalog" +weight: 11 +type: docs +aliases: + - /dev/table/connectors/glue.html +--- +<!-- +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. +--> + +# AWS Glue Catalog + +The AWS Glue Catalog provides a way to use [AWS Glue](https://aws.amazon.com/glue) as a catalog for Apache Flink. +This allows users to access Glue's metadata store directly from Flink SQL and Table API. Review Comment: In my understanding, the DataStream API can also utilize the Glue Catalog. Is that not true? ########## flink-catalog-aws/flink-catalog-aws-glue/src/main/java/org/apache/flink/table/catalog/glue/factory/GlueCatalogFactory.java: ########## @@ -0,0 +1,83 @@ +/* + * 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.flink.table.catalog.glue.factory; + +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.glue.GlueCatalog; +import org.apache.flink.table.factories.CatalogFactory; + +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +/** + * Factory for creating GlueCatalog instances. + */ +public class GlueCatalogFactory implements CatalogFactory { + + // Define configuration options that users must provide + public static final ConfigOption<String> REGION = + ConfigOptions.key("region") + .stringType() + .noDefaultValue() + .withDescription("AWS region for the Glue catalog"); + + public static final ConfigOption<String> DEFAULT_DATABASE = + ConfigOptions.key("default-database") + .stringType() + .defaultValue("default") + .withDescription("Default database to use in Glue catalog"); + + @Override + public String factoryIdentifier() { + return "glue"; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + Set<ConfigOption<?>> options = new HashSet<>(); + options.add(REGION); + return options; + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + Set<ConfigOption<?>> options = new HashSet<>(); + options.add(DEFAULT_DATABASE); + return options; + } + + @Override + public Catalog createCatalog(Context context) { + Map<String, String> config = context.getOptions(); + String name = context.getName(); + String region = config.get(REGION.key()); + String defaultDatabase = config.getOrDefault(DEFAULT_DATABASE.key(), DEFAULT_DATABASE.defaultValue()); + + // Ensure required properties are present Review Comment: unwanted comment ########## flink-catalog-aws/flink-catalog-aws-glue/src/main/java/org/apache/flink/table/catalog/glue/operator/GlueFunctionOperator.java: ########## @@ -0,0 +1,295 @@ +/* + * 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.flink.table.catalog.glue.operator; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.catalog.CatalogFunction; +import org.apache.flink.table.catalog.CatalogFunctionImpl; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.glue.util.GlueCatalogConstants; +import org.apache.flink.table.catalog.glue.util.GlueFunctionsUtil; +import org.apache.flink.table.resource.ResourceUri; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.glue.model.AlreadyExistsException; +import software.amazon.awssdk.services.glue.model.CreateUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.CreateUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.DeleteUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.DeleteUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.EntityNotFoundException; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionsRequest; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionsResponse; +import software.amazon.awssdk.services.glue.model.GlueException; +import software.amazon.awssdk.services.glue.model.PrincipalType; +import software.amazon.awssdk.services.glue.model.UpdateUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.UpdateUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.UserDefinedFunction; +import software.amazon.awssdk.services.glue.model.UserDefinedFunctionInput; + +import java.util.Collection; +import java.util.LinkedList; +import java.util.List; +import java.util.Optional; +import java.util.stream.Collectors; + +/** Utilities for Glue catalog Function related operations. */ +@Internal +public class GlueFunctionOperator extends GlueOperator { + + private static final Logger LOG = LoggerFactory.getLogger(GlueFunctionOperator.class); + + /** + * Constructor to initialize the shared fields. + * + * @param glueClient The Glue client used for interacting with the AWS Glue service. + * @param catalogName The catalog name associated with the Glue operations. + */ + public GlueFunctionOperator(GlueClient glueClient, String catalogName) { + super(glueClient, catalogName); + } + + /** + * Create a function. Function name should be handled in a case-insensitive way. + * + * @param functionPath path of the function + * @param function Flink function to be created + * @throws CatalogException in case of any runtime exception + */ + public void createGlueFunction(ObjectPath functionPath, CatalogFunction function) + throws CatalogException, FunctionAlreadyExistException { + UserDefinedFunctionInput functionInput = createFunctionInput(functionPath, function); + CreateUserDefinedFunctionRequest.Builder createUDFRequest = + CreateUserDefinedFunctionRequest.builder() + .databaseName(functionPath.getDatabaseName()) + .functionInput(functionInput); + try { + CreateUserDefinedFunctionResponse response = + glueClient.createUserDefinedFunction(createUDFRequest.build()); + if (response == null || (response.sdkHttpResponse() != null && !response.sdkHttpResponse().isSuccessful())) { + throw new CatalogException("Error creating function: " + functionPath.getFullName()); + } + LOG.info("Created Function: {}", functionPath.getFullName()); + } catch (AlreadyExistsException e) { + LOG.error( + String.format( + "%s already Exists. Function language of type: %s. \n%s", + functionPath.getFullName(), function.getFunctionLanguage(), e)); + throw new FunctionAlreadyExistException(catalogName, functionPath, e); + } catch (GlueException e) { + LOG.error("Error creating glue function: {}\n{}", functionPath.getFullName(), e); + throw new CatalogException(GlueCatalogConstants.GLUE_EXCEPTION_MSG_IDENTIFIER, e); + } + } + + /** + * Modify an existing function. Function name should be handled in a case-insensitive way. + * + * @param functionPath path of function. + * @param newFunction modified function. + * @throws CatalogException on runtime errors. + * @throws FunctionNotExistException if the function doesn't exist. + */ + public void alterGlueFunction(ObjectPath functionPath, CatalogFunction newFunction) + throws CatalogException, FunctionNotExistException { + + UserDefinedFunctionInput functionInput = createFunctionInput(functionPath, newFunction); + + UpdateUserDefinedFunctionRequest updateUserDefinedFunctionRequest = + UpdateUserDefinedFunctionRequest.builder() + .functionName(functionPath.getObjectName()) + .databaseName(functionPath.getDatabaseName()) + .functionInput(functionInput) + .build(); + try { + UpdateUserDefinedFunctionResponse response = + glueClient.updateUserDefinedFunction(updateUserDefinedFunctionRequest); + if (response == null || (response.sdkHttpResponse() != null && !response.sdkHttpResponse().isSuccessful())) { + throw new CatalogException("Error altering function: " + functionPath.getFullName()); + } + LOG.info("Altered Function: {}", functionPath.getFullName()); + } catch (EntityNotFoundException e) { + LOG.error("Function not found: {}", functionPath.getFullName()); + throw new FunctionNotExistException(catalogName, functionPath, e); + } catch (GlueException e) { + LOG.error("Error altering glue function: {}\n{}", functionPath.getFullName(), e); + throw new CatalogException(GlueCatalogConstants.GLUE_EXCEPTION_MSG_IDENTIFIER, e); + } + } + + /** + * Get the user defined function from glue Catalog. Function name should be handled in a + * case-insensitive way. + * + * @param functionPath path of the function + * @return the requested function + * @throws CatalogException in case of any runtime exception + * @throws FunctionNotExistException if the function doesn't exist + */ + public CatalogFunction getGlueFunction(ObjectPath functionPath) throws CatalogException, FunctionNotExistException { + GetUserDefinedFunctionRequest request = + GetUserDefinedFunctionRequest.builder() + .databaseName(functionPath.getDatabaseName()) + .functionName(functionPath.getObjectName()) + .build(); + try { + GetUserDefinedFunctionResponse response = glueClient.getUserDefinedFunction(request); + UserDefinedFunction udf = response.userDefinedFunction(); + List<ResourceUri> resourceUris = + udf.resourceUris().stream() + .map( + resourceUri -> + new org.apache.flink.table.resource.ResourceUri( + org.apache.flink.table.resource.ResourceType + .valueOf(resourceUri.resourceType().name()), + resourceUri.uri())) + .collect(Collectors.toList()); + return new CatalogFunctionImpl( + GlueFunctionsUtil.getCatalogFunctionClassName(udf), + GlueFunctionsUtil.getFunctionalLanguage(udf), + resourceUris); + } catch (EntityNotFoundException e) { + LOG.error("Function not found: {}", functionPath.getFullName()); + throw new FunctionNotExistException(catalogName, functionPath, e); + } catch (GlueException e) { + LOG.error("Error fetching function {}: {}", functionPath.getFullName(), e); + throw new CatalogException( + String.format("Error getting function %s: %s", functionPath.getFullName(), e.getMessage()), e); + } + } + + public List<String> listGlueFunctions(String databaseName) { + GetUserDefinedFunctionsRequest.Builder functionsRequest = + GetUserDefinedFunctionsRequest.builder() + .databaseName(databaseName); + List<String> glueFunctions; Review Comment: If there is no function present, then `glueFunctions` will be a null object. This might be an issue. can you add ```suggestion List<String> glueFunctions = new ArrayList<>(); ``` ########## flink-catalog-aws/flink-catalog-aws-glue/src/main/java/org/apache/flink/table/catalog/glue/operator/GlueOperator.java: ########## @@ -0,0 +1,45 @@ +/* + * 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.flink.table.catalog.glue.operator; + +import software.amazon.awssdk.services.glue.GlueClient; + +/** + * Abstract base class for Glue operations that contains common functionality + * for interacting with the AWS Glue service. + */ +public abstract class GlueOperator { + + /** The Glue client used for interacting with AWS Glue. */ + protected final GlueClient glueClient; + + /** The catalog name associated with the Glue operations. */ + protected final String catalogName; + + /** + * Constructor to initialize the shared fields. + * + * @param glueClient The Glue client used for interacting with the AWS Glue service. + * @param catalogName The catalog name associated with the Glue operations. + */ + protected GlueOperator(GlueClient glueClient, String catalogName) { Review Comment: Add a null check on input params `glueClient` -- 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. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
