You are viewing a plain text version of this content. The canonical link for it is here.

Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2021/08/10 20:58:12 UTC

Change in asterixdb[master]: [ASTERIXDB-2945][EXT] Ensure the size of S3 connection pool

From Wael Alkowaileet <wa...@gmail.com>:

Wael Alkowaileet has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/12763 )


Change subject: [ASTERIXDB-2945][EXT] Ensure the size of S3 connection pool
......................................................................

[ASTERIXDB-2945][EXT] Ensure the size of S3 connection pool

- user mode changes: no
- storage format changes: no
- interface changes: no

Details:
- We set the S3 connection pool size to twice the number
  of partitions.

Change-Id: I1e1ce66cc7cd39cc81d004f90c36871ad31a685f
---
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
3 files changed, 48 insertions(+), 29 deletions(-)



  git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/63/12763/1

diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
index c9eb489..803e657 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
@@ -51,7 +51,8 @@
 
         //Configure Hadoop S3 input splits
         JobConf conf = createHdfsConf(serviceCtx, configuration);
-        ExternalDataUtils.AwsS3.configureAwsS3HdfsJobConf(conf, configuration);
+        int numberOfPartitions = getPartitionConstraint().getLocations().length;
+        ExternalDataUtils.AwsS3.configureAwsS3HdfsJobConf(conf, configuration, numberOfPartitions);
         configureHdfsConf(conf, configuration);
     }
 
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
index 51865f0..7445fbd 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
@@ -320,9 +320,17 @@
         public static final String HADOOP_ACCESS_KEY_ID = "fs.s3a.access.key";
         public static final String HADOOP_SECRET_ACCESS_KEY = "fs.s3a.secret.key";
         public static final String HADOOP_SESSION_TOKEN = "fs.s3a.session.token";
-        public static final String HADOOP_PATH_STYLE_ACCESS = "fs.s3a.path.style.access";
         public static final String HADOOP_REGION = "fs.s3a.region";
         public static final String HADOOP_SERVICE_END_POINT = "fs.s3a.endpoint";
+
+        /*
+         * Internal configurations
+         */
+        //Allows accessing directories as file system path
+        public static final String HADOOP_PATH_STYLE_ACCESS = "fs.s3a.path.style.access";
+        //The number of maximum HTTP connections in connection pool
+        public static final String HADOOP_S3_CONNECTION_POOL_SIZE = "fs.s3a.connection.maximum";
+        //S3 used protocol
         public static final String HADOOP_S3_PROTOCOL = "s3a";
 
         //Hadoop credentials provider key
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
index 750ff13..fbdcd30 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
@@ -25,6 +25,7 @@
 import static org.apache.asterix.external.util.ExternalDataConstants.AwsS3.HADOOP_ANONYMOUS_ACCESS;
 import static org.apache.asterix.external.util.ExternalDataConstants.AwsS3.HADOOP_CREDENTIAL_PROVIDER_KEY;
 import static org.apache.asterix.external.util.ExternalDataConstants.AwsS3.HADOOP_PATH_STYLE_ACCESS;
+import static org.apache.asterix.external.util.ExternalDataConstants.AwsS3.HADOOP_S3_CONNECTION_POOL_SIZE;
 import static org.apache.asterix.external.util.ExternalDataConstants.AwsS3.HADOOP_SECRET_ACCESS_KEY;
 import static org.apache.asterix.external.util.ExternalDataConstants.AwsS3.HADOOP_SESSION_TOKEN;
 import static org.apache.asterix.external.util.ExternalDataConstants.AwsS3.HADOOP_TEMP_ACCESS;
@@ -96,6 +97,8 @@
 import org.apache.hyracks.dataflow.common.data.parsers.LongParserFactory;
 import org.apache.hyracks.dataflow.common.data.parsers.UTF8StringParserFactory;
 import org.apache.hyracks.util.StorageUtil;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import com.azure.storage.blob.BlobContainerClient;
 import com.azure.storage.blob.BlobServiceClient;
@@ -121,7 +124,7 @@
 import software.amazon.awssdk.services.s3.model.S3Response;
 
 public class ExternalDataUtils {
-
+    private static final Logger LOGGER = LogManager.getLogger();
     private static final Map<ATypeTag, IValueParserFactory> valueParserFactoryMap = new EnumMap<>(ATypeTag.class);
     private static final int DEFAULT_MAX_ARGUMENT_SZ = 1024 * 1024;
     private static final int HEADER_FUDGE = 64;
@@ -196,8 +199,8 @@
     }
 
     public static boolean isExternal(String aString) {
-        return ((aString != null) && aString.contains(ExternalDataConstants.EXTERNAL_LIBRARY_SEPARATOR)
-                && (aString.trim().length() > 1));
+        return ((aString != null) && aString.contains(ExternalDataConstants.EXTERNAL_LIBRARY_SEPARATOR) && (
+                aString.trim().length() > 1));
     }
 
     public static String getLibraryName(String aString) {
@@ -284,7 +287,8 @@
             throw new AsterixException("to use " + ExternalDataConstants.EXTERNAL + " reader, the parameter "
                     + ExternalDataConstants.KEY_READER_FACTORY + " must be specified.");
         }
-        String[] libraryAndFactory = readerFactory.split(ExternalDataConstants.EXTERNAL_LIBRARY_SEPARATOR); //TODO(MULTI_PART_DATAVERSE_NAME):REVISIT
+        String[] libraryAndFactory = readerFactory.split(
+                ExternalDataConstants.EXTERNAL_LIBRARY_SEPARATOR); //TODO(MULTI_PART_DATAVERSE_NAME):REVISIT
         if (libraryAndFactory.length != 2) {
             throw new AsterixException("The parameter " + ExternalDataConstants.KEY_READER_FACTORY
                     + " must follow the format \"DataverseName.LibraryName#ReaderFactoryFullyQualifiedName\"");
@@ -294,7 +298,8 @@
             throw new AsterixException("The parameter " + ExternalDataConstants.KEY_READER_FACTORY
                     + " must follow the format \"DataverseName.LibraryName#ReaderFactoryFullyQualifiedName\"");
         }
-        DataverseName dataverseName = DataverseName.createSinglePartName(dataverseAndLibrary[0]); //TODO(MULTI_PART_DATAVERSE_NAME):REVISIT
+        DataverseName dataverseName =
+                DataverseName.createSinglePartName(dataverseAndLibrary[0]); //TODO(MULTI_PART_DATAVERSE_NAME):REVISIT
         String libraryName = dataverseAndLibrary[1];
         ILibrary lib;
         try {
@@ -329,10 +334,8 @@
                 throw new AsterixException("Unexpected library language: " + lib.getLanguage());
             }
             ClassLoader classLoader = ((JavaLibrary) lib).getClassLoader();
-            return (IDataParserFactory) classLoader
-                    .loadClass(parserFactoryName
-                            .substring(parserFactoryName.indexOf(ExternalDataConstants.EXTERNAL_LIBRARY_SEPARATOR) + 1))
-                    .newInstance();
+            return (IDataParserFactory) classLoader.loadClass(parserFactoryName.substring(
+                    parserFactoryName.indexOf(ExternalDataConstants.EXTERNAL_LIBRARY_SEPARATOR) + 1)).newInstance();
         } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) {
             throw new AsterixException("Failed to create an external parser factory", e);
         }
@@ -455,8 +458,8 @@
             configuration.put(ExternalDataConstants.KEY_PARSER, ExternalDataConstants.FORMAT_NOOP);
             configuration.put(ExternalDataConstants.KEY_FORMAT, ExternalDataConstants.FORMAT_PARQUET);
         }
-        if (!configuration.containsKey(ExternalDataConstants.KEY_PARSER)
-                && configuration.containsKey(ExternalDataConstants.KEY_FORMAT)) {
+        if (!configuration.containsKey(ExternalDataConstants.KEY_PARSER) && configuration.containsKey(
+                ExternalDataConstants.KEY_FORMAT)) {
             configuration.put(ExternalDataConstants.KEY_PARSER, configuration.get(ExternalDataConstants.KEY_FORMAT));
         }
     }
@@ -652,8 +655,9 @@
                         String stuffEscaped = stuffBuilder.toString();
 
                         // Escape the set operations
-                        stuffEscaped = stuffEscaped.replace("&&", "\\&\\&").replace("~~", "\\~\\~")
-                                .replace("||", "\\|\\|").replace("--", "\\-\\-");
+                        stuffEscaped =
+                                stuffEscaped.replace("&&", "\\&\\&").replace("~~", "\\~\\~").replace("||", "\\|\\|")
+                                        .replace("--", "\\-\\-");
 
                         result.append("[").append(stuffEscaped).append("]");
                         charPosition = closingBracketPosition + 1;
@@ -701,19 +705,17 @@
                 includes.add(entry);
             } else if (key.equals(ExternalDataConstants.KEY_EXCLUDE)) {
                 excludes.add(entry);
-            } else if (key.startsWith(ExternalDataConstants.KEY_INCLUDE)
-                    || key.startsWith(ExternalDataConstants.KEY_EXCLUDE)) {
+            } else if (key.startsWith(ExternalDataConstants.KEY_INCLUDE) || key.startsWith(
+                    ExternalDataConstants.KEY_EXCLUDE)) {
 
                 // Split by the "#", length should be 2, left should be include/exclude, right should be integer
                 String[] splits = key.split("#");
 
-                if (key.startsWith(ExternalDataConstants.KEY_INCLUDE) && splits.length == 2
-                        && splits[0].equals(ExternalDataConstants.KEY_INCLUDE)
-                        && NumberUtils.isIntegerNumericString(splits[1])) {
+                if (key.startsWith(ExternalDataConstants.KEY_INCLUDE) && splits.length == 2 && splits[0].equals(
+                        ExternalDataConstants.KEY_INCLUDE) && NumberUtils.isIntegerNumericString(splits[1])) {
                     includes.add(entry);
-                } else if (key.startsWith(ExternalDataConstants.KEY_EXCLUDE) && splits.length == 2
-                        && splits[0].equals(ExternalDataConstants.KEY_EXCLUDE)
-                        && NumberUtils.isIntegerNumericString(splits[1])) {
+                } else if (key.startsWith(ExternalDataConstants.KEY_EXCLUDE) && splits.length == 2 && splits[0].equals(
+                        ExternalDataConstants.KEY_EXCLUDE) && NumberUtils.isIntegerNumericString(splits[1])) {
                     excludes.add(entry);
                 } else {
                     throw new CompilationException(ErrorCode.INVALID_PROPERTY_FORMAT, key);
@@ -801,8 +803,8 @@
             } else {
                 // auth required, check for temporary or permanent credentials
                 if (sessionToken != null) {
-                    credentialsProvider = StaticCredentialsProvider
-                            .create(AwsSessionCredentials.create(accessKeyId, secretAccessKey, sessionToken));
+                    credentialsProvider = StaticCredentialsProvider.create(
+                            AwsSessionCredentials.create(accessKeyId, secretAccessKey, sessionToken));
                 } else {
                     credentialsProvider =
                             StaticCredentialsProvider.create(AwsBasicCredentials.create(accessKeyId, secretAccessKey));
@@ -833,9 +835,11 @@
         /**
          * Builds the S3 client using the provided configuration
          *
-         * @param configuration properties
+         * @param configuration      properties
+         * @param numberOfPartitions number of partitions in the cluster
          */
-        public static void configureAwsS3HdfsJobConf(JobConf conf, Map<String, String> configuration) {
+        public static void configureAwsS3HdfsJobConf(JobConf conf, Map<String, String> configuration,
+                int numberOfPartitions) {
             String accessKeyId = configuration.get(ExternalDataConstants.AwsS3.ACCESS_KEY_ID_FIELD_NAME);
             String secretAccessKey = configuration.get(ExternalDataConstants.AwsS3.SECRET_ACCESS_KEY_FIELD_NAME);
             String sessionToken = configuration.get(ExternalDataConstants.AwsS3.SESSION_TOKEN_FIELD_NAME);
@@ -867,6 +871,11 @@
              */
             conf.set(HADOOP_PATH_STYLE_ACCESS, ExternalDataConstants.TRUE);
 
+            /*
+             * Set the size of S3 connection pool to twice the number of partitions
+             */
+            conf.set(HADOOP_S3_CONNECTION_POOL_SIZE, String.valueOf(2 * numberOfPartitions));
+
             if (serviceEndpoint != null) {
                 // Validation of the URL should be done at hadoop-aws level
                 conf.set(ExternalDataConstants.AwsS3.HADOOP_SERVICE_END_POINT, serviceEndpoint);
@@ -935,8 +944,9 @@
                 }
             }
 
-            boolean isEmpty = useOldApi ? ((ListObjectsResponse) response).contents().isEmpty()
-                    : ((ListObjectsV2Response) response).contents().isEmpty();
+            boolean isEmpty = useOldApi ?
+                    ((ListObjectsResponse) response).contents().isEmpty() :
+                    ((ListObjectsV2Response) response).contents().isEmpty();
             if (isEmpty && collector.shouldWarn()) {
                 Warning warning = Warning.of(srcLoc, ErrorCode.EXTERNAL_SOURCE_CONFIGURATION_RETURNED_NO_FILES);
                 collector.warn(warning);

-- 
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/12763
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings

Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I1e1ce66cc7cd39cc81d004f90c36871ad31a685f
Gerrit-Change-Number: 12763
Gerrit-PatchSet: 1
Gerrit-Owner: Wael Alkowaileet <wa...@gmail.com>
Gerrit-MessageType: newchange