Merge pull request #200 from marklogic/feature/total-thread-count

Supporting --thread-count and --thread-count-per-partition now

Author: rjrudin

docs/import/import-files/aggregate-xml.md

@@ -46,7 +46,8 @@ of the `person` element in the document:
 ./bin/flux import-aggregate-xml-files \
     --path /data/people.xml \
     --connection-string user:password@localhost:8000 \
-    --element employee --namespace org:example
+    --element person \
+    --namespace org:example
 ```
 
 ## Controlling document URIs
@@ -60,8 +61,8 @@ based on the value of each `id` element in the `org:example` namespace:
 ./bin/flux import-aggregate-xml-files \
     --path /data/people.xml \
     --connection-string user:password@localhost:8000 \
-    --element employee --namespace org:example \
-    --uri-element id --namespace org:example
+    --element person --namespace org:example \
+    --uri-element id --uri-namespace org:example
 ```
 
 You may still wish to use options like `--uri-prefix` and `--uri-suffix` to make the URI more self-describing. For

docs/import/tuning-performance.md

@@ -13,10 +13,9 @@ Batch size is configured via the `--batch-size` option, which defaults to a valu
 your documents, you may find improved performance by raising this value significantly for smaller documents, such as 500
 or even 1000.
 
-For the number of threads used to send requests to MarkLogic, two factors come into play. The product of the
-number of partitions and the value of the `--thread-count` option determines how many total threads will be used to send
-requests. For example, if the import command uses 4 partitions to read data and `--thread-count` is set to 4 (its
-default value), 16 total threads will send requests to MarkLogic.
+For the number of threads used to send requests to MarkLogic, two factors come into play.
+The value of the `--thread-count` option determines how many total threads will be used across all partitions to send
+requests. You can alternatively configure a number of threads per partition using `--thread-count-per-partition`.
 
 The number of partitions is determined by how data is read and differs across the various import commands.
 Flux will log the number of partitions for each import command as shown below:
@@ -35,7 +34,7 @@ by placing a load balancer in front of MarkLogic or by configuring direct connec
 
 The rule of thumb can thus be expressed as:
 
-    Number of partitions * Value of --thread-count <= Number of hosts * number of app server threads
+    Value of --thread-count <= Number of hosts * number of app server threads
 
 ### Direct connections to each host
 

flux-cli/src/main/java/com/marklogic/flux/api/WriteDocumentsOptions.java

@@ -23,7 +23,7 @@ public interface WriteDocumentsOptions<T extends WriteDocumentsOptions> {
 
     T threadCount(int threadCount);
 
-    T totalThreadCount(int totalThreadCount);
+    T threadCountPerPartition(int threadCountPerPartition);
 
     T transform(String transform);
 

flux-cli/src/main/java/com/marklogic/flux/impl/copy/CopyCommand.java

@@ -98,15 +98,16 @@ public static class CopyWriteDocumentsParams implements WriteDocumentsOptions<Co
 
         @CommandLine.Option(
             names = "--output-thread-count",
-            description = "The number of threads used by each partition worker when writing batches of documents to MarkLogic."
+            description = "The total number of threads used across all partitions when writing batches of documents to MarkLogic."
         )
         private int threadCount = 4;
 
         @CommandLine.Option(
-            names = "--output-total-thread-count",
-            description = "The total number of threads used across all partitions when writing batches of documents to MarkLogic."
+            names = "--output-thread-count-per-partition",
+            description = "The number of threads used by each partition worker when writing batches of documents to MarkLogic. " +
+                "Takes precedence over the '--output-thread-count' option."
         )
-        private int totalThreadCount;
+        private int threadCountPerPartition;
 
         @CommandLine.Option(
             names = "--output-transform",
@@ -163,7 +164,7 @@ protected Map<String, String> makeOptions() {
                 Options.WRITE_PERMISSIONS, permissions,
                 Options.WRITE_TEMPORAL_COLLECTION, temporalCollection,
                 Options.WRITE_THREAD_COUNT, OptionsUtil.intOption(threadCount),
-                Options.WRITE_TOTAL_THREAD_COUNT, OptionsUtil.intOption(totalThreadCount),
+                Options.WRITE_THREAD_COUNT_PER_PARTITION, OptionsUtil.intOption(threadCountPerPartition),
                 Options.WRITE_TRANSFORM_NAME, transform,
                 Options.WRITE_TRANSFORM_PARAMS, transformParams,
                 Options.WRITE_TRANSFORM_PARAMS_DELIMITER, transformParamsDelimiter,
@@ -229,8 +230,8 @@ public CopyWriteDocumentsParams threadCount(int threadCount) {
         }
 
         @Override
-        public CopyWriteDocumentsParams totalThreadCount(int totalThreadCount) {
-            this.totalThreadCount = totalThreadCount;
+        public CopyWriteDocumentsParams threadCountPerPartition(int threadCountPerPartition) {
+            this.threadCountPerPartition = threadCountPerPartition;
             return this;
         }
 

flux-cli/src/main/java/com/marklogic/flux/impl/importdata/WriteDocumentParams.java

@@ -64,16 +64,16 @@ public class WriteDocumentParams<T extends WriteDocumentsOptions> implements Wri
 
     @CommandLine.Option(
         names = "--thread-count",
-        description = "The number of threads used by each partition worker when writing batches of documents to MarkLogic."
+        description = "The total number of threads used across all partitions when writing batches of documents to MarkLogic."
     )
     private int threadCount = 4;
 
     @CommandLine.Option(
-        names = "--total-thread-count",
-        description = "The total number of threads used across all partitions when writing batches of documents to MarkLogic. " +
+        names = "--thread-count-per-partition",
+        description = "The number of threads used by each partition worker when writing batches of documents to MarkLogic. " +
             "Takes precedence over the '--thread-count' option."
     )
-    private int totalThreadCount;
+    private int threadCountPerPartition;
 
     @CommandLine.Option(
         names = "--transform",
@@ -130,7 +130,7 @@ public Map<String, String> makeOptions() {
             Options.WRITE_PERMISSIONS, permissions,
             Options.WRITE_TEMPORAL_COLLECTION, temporalCollection,
             Options.WRITE_THREAD_COUNT, OptionsUtil.intOption(threadCount),
-            Options.WRITE_TOTAL_THREAD_COUNT, OptionsUtil.intOption(totalThreadCount),
+            Options.WRITE_THREAD_COUNT_PER_PARTITION, OptionsUtil.intOption(threadCountPerPartition),
             Options.WRITE_TRANSFORM_NAME, transform,
             Options.WRITE_TRANSFORM_PARAMS, transformParams,
             Options.WRITE_TRANSFORM_PARAMS_DELIMITER, transformParamsDelimiter,
@@ -201,8 +201,8 @@ public T threadCount(int threadCount) {
     }
 
     @Override
-    public T totalThreadCount(int totalThreadCount) {
-        this.totalThreadCount = totalThreadCount;
+    public T threadCountPerPartition(int threadCountPerPartition) {
+        this.threadCountPerPartition = threadCountPerPartition;
         return (T) this;
     }
 

flux-cli/src/main/resources/marklogic-spark-messages_en.properties

@@ -14,7 +14,7 @@ spark.marklogic.write.graph=--graph
 spark.marklogic.write.graphOverride=--graph-override
 spark.marklogic.write.jsonRootName=--json-root-name
 spark.marklogic.write.threadCount=--thread-count
-spark.marklogic.write.totalThreadCount=--total-thread-count
+spark.marklogic.write.threadCountPerPartition=--thread-count-per-partition
 spark.marklogic.write.transformParams=--transform-params
 spark.marklogic.write.uriTemplate=--uri-template
 spark.marklogic.write.xmlRootName=--xml-root-name

flux-cli/src/test/java/com/marklogic/flux/impl/ErrorMessagesTest.java

@@ -30,7 +30,7 @@ void verifyEachKeyIsOverridden() {
         assertEquals("--graph-override", bundle.getString(Options.WRITE_GRAPH_OVERRIDE));
         assertEquals("--json-root-name", bundle.getString(Options.WRITE_JSON_ROOT_NAME));
         assertEquals("--thread-count", bundle.getString(Options.WRITE_THREAD_COUNT));
-        assertEquals("--total-thread-count", bundle.getString(Options.WRITE_TOTAL_THREAD_COUNT));
+        assertEquals("--thread-count-per-partition", bundle.getString(Options.WRITE_THREAD_COUNT_PER_PARTITION));
         assertEquals("--transform-params", bundle.getString(Options.WRITE_TRANSFORM_PARAMS));
         assertEquals("--uri-template", bundle.getString(Options.WRITE_URI_TEMPLATE));
         assertEquals("--xml-root-name", bundle.getString(Options.WRITE_XML_ROOT_NAME));

flux-cli/src/test/java/com/marklogic/flux/impl/importdata/ImportJdbcTest.java

@@ -82,7 +82,7 @@ void allCustomers() {
             "--collections", "customer",
             "--repartition", "2",
             // Just verifying that these work without causing any errors.
-            "--total-thread-count", "16",
+            "--thread-count-per-partition", "8",
             "--batch-size", "10"
         );
 

flux-cli/src/test/java/com/marklogic/flux/impl/importdata/ImportParquetFilesTest.java

@@ -21,7 +21,7 @@ void defaultSettingsSingleFile() {
             "--connection-string", makeConnectionString(),
             "--permissions", DEFAULT_PERMISSIONS,
             "--collections", "parquet-test",
-            "--total-thread-count", "1",
+            "--thread-count", "1",
             "--uri-template", "/parquet/{model}.json",
 
             // Including these for manual verification of progress logging.

mlcp-testing/build.gradle

@@ -81,9 +81,9 @@ task mlcpImportRdf(type: Exec) {
     "-output_permissions", "rest-reader,read,rest-writer,update"
 }
 
-task ntImportRdf(type: Exec) {
+task fluxImportRdf(type: Exec) {
   description = "Intended to run against a local Caddy load balancer."
-  workingDir = "../nt/"
+  workingDir = "../flux/"
   commandLine "./bin/flux", "import-rdf-files",
     "--connection-string", "${mlUsername}:${mlPassword}@${lbHost}:${lbPort}",
     "--path", rdfFile,
@@ -93,9 +93,9 @@ task ntImportRdf(type: Exec) {
     "--batch-size", "100"
 }
 
-task ntDirectImportRdf(type: Exec) {
+task fluxDirectImportRdf(type: Exec) {
   description = "For testing without a load balancer."
-  workingDir = "../nt/"
+  workingDir = "../flux/"
   commandLine "./bin/flux", "import-rdf-files",
     "--connection-string", "${mlUsername}:${mlPassword}@${mlHost}:${mlRestPort}",
     "--path", rdfFile,
@@ -105,8 +105,8 @@ task ntDirectImportRdf(type: Exec) {
     "--batch-size", "100"
 }
 
-task ntCopy(type: Exec) {
-  workingDir = "../nt/"
+task fluxCopy(type: Exec) {
+  workingDir = "../flux/"
   commandLine "./bin/flux", "copy",
     "--connection-string", "${mlUsername}:${mlPassword}@${mlHost}:${mlRestPort}",
     "--collections", "address_small",
@@ -154,11 +154,11 @@ task mlcpExportArchive(type: Exec) {
     "-collection_filter", "author"
 }
 
-task ntTwoWaySSL(type: Exec) {
+task fluxTwoWaySSL(type: Exec) {
   description = "For manual testing with the Java Client's java-unittest app server when its TwoWaySSLTest has been " +
     "paused in a debugger. You can copy the keyStore/trustStore files generated by that test to this directory and " +
     "then run this task to ensure that two-way SSL works."
-  workingDir = "../nt/"
+  workingDir = "../flux/"
   commandLine "./bin/flux", "copy",
     "--connection-string", "rest-writer:x@localhost:8012",
     "--collections", "zipcode",