rm the useless write_buffer_size for data portal worker (bytedance#184)

* rm the useless write_buffer_size for data portal worker * make the web_console ut pass * fix the web_console ut pass Co-authored-by: fangchenliaohui <[email protected]>
zhangqixun · Jul 26, 2020 · 7752696 · 7752696
1 parent 3fbaf7a
commit 7752696
Show file tree

Hide file tree

Showing 9 changed files with 7 additions and 18 deletions.
diff --git a/deploy/scripts/data_portal/run_data_portal_worker.sh b/deploy/scripts/data_portal/run_data_portal_worker.sh
@@ -23,7 +23,6 @@ source /app/deploy/scripts/env_to_args.sh
 MASTER_POD_NAMES=`python -c 'import json, os; print(json.loads(os.environ["CLUSTER_SPEC"])["clusterSpec"]["Master"][0])'`
 
 merge_buffer_size=$(normalize_env_to_args "--merge_buffer_size" $MERGE_BUFFER_SIZE)
-writer_buffer_size=$(normalize_env_to_args "--write_buffer_size" $WRITE_BUFFER_SIZE)
 input_data_file_iter=$(normalize_env_to_args "--input_data_file_iter" $INPUT_DATA_FORMAT)
 compressed_type=$(normalize_env_to_args "--compressed_type" $COMPRESSED_TYPE)
 output_builder=$(normalize_env_to_args "--output_builder" $OUTPUT_DATA_FORMAT)
@@ -38,7 +37,7 @@ python -m fedlearner.data_join.cmd.data_portal_worker_cli \
   --etcd_name=$ETCD_NAME \
   --etcd_addrs=$ETCD_ADDR \
   --etcd_base_dir=$ETCD_BASE_DIR \
-  $merge_buffer_size $write_buffer_size \
+  $merge_buffer_size \
   $input_data_file_iter $compressed_type \
   $output_builder $builder_compressed_type \
   $batch_size $max_flying_item $USE_MOCK_ETCD

diff --git a/fedlearner/data_join/cmd/data_portal_worker_cli.py b/fedlearner/data_join/cmd/data_portal_worker_cli.py
@@ -39,8 +39,6 @@
                         help='use to mock etcd for test')
     parser.add_argument("--merge_buffer_size", type=int,
                         default=4096, help="the buffer size for merging")
-    parser.add_argument("--write_buffer_size", type=int, default=10485760,
-                        help="the output buffer size (bytes) for partitioner")
     parser.add_argument("--merger_read_ahead_size", type=int, default=0,
                         help="the read ahead size for merger")
     parser.add_argument("--input_data_file_iter", type=str, default="TF_RECORD",
@@ -75,7 +73,6 @@
             max_flying_item=args.max_flying_item
         ),
         merge_buffer_size=args.merge_buffer_size,
-        write_buffer_size=args.write_buffer_size,
         merger_read_ahead_size=args.merger_read_ahead_size
     )
 

diff --git a/fedlearner/data_join/data_portal_worker.py b/fedlearner/data_join/data_portal_worker.py
@@ -150,8 +150,7 @@ def _make_merger_options(self, task):
             writer_options=self._options.writer_options,
             output_file_dir=task.reduce_base_dir,
             partition_id=task.partition_id,
-            merge_buffer_size=self._options.merge_buffer_size,
-            write_buffer_size=self._options.write_buffer_size
+            merge_buffer_size=self._options.merge_buffer_size
         )
 
     def _run_map_task(self, task):

diff --git a/protocols/fedlearner/common/data_join_service.proto b/protocols/fedlearner/common/data_join_service.proto
@@ -330,5 +330,4 @@ message SortRunMergerOptions {
   string output_file_dir = 4;
   int64 partition_id = 5;
   int64 merge_buffer_size = 6;
-  int64 write_buffer_size = 7;
 }
diff --git a/protocols/fedlearner/common/data_portal_service.proto b/protocols/fedlearner/common/data_portal_service.proto
@@ -112,7 +112,6 @@ message DataPortalWorkerOptions {
   WriterOptions writer_options = 2;
   BatchProcessorOptions batch_processor_options = 3;
   int64 merge_buffer_size = 4;
-  int64 write_buffer_size = 5;
-  int64 merger_read_ahead_size = 6;
-  bool use_mock_etcd = 7;
+  int64 merger_read_ahead_size = 5;
+  bool use_mock_etcd = 6;
 }
diff --git a/web_console/tests/fixtures/test_data_join.yaml b/web_console/tests/fixtures/test_data_join.yaml
@@ -28,7 +28,7 @@ spec:
                   fieldRef:
                     fieldPath: metadata.name
               - name: ROLE
-                value: "Follower"
+                value: "follower"
               - name: APPLICATION_ID
                 value: test_data_join
               - name: OUTPUT_BASE_DIR
@@ -96,7 +96,7 @@ spec:
                   fieldRef:
                     fieldPath: metadata.name
               - name: ROLE
-                value: "Follower"
+                value: "follower"
               - name: APPLICATION_ID
                 value: test_data_join
               - name: OUTPUT_BASE_DIR

diff --git a/web_console/tests/fixtures/test_data_portal.yaml b/web_console/tests/fixtures/test_data_portal.yaml
@@ -42,7 +42,7 @@ spec:
               - name: INPUT_BASE_DIR
                 value: /data/criteo_data/100wexamples
               - name: OUTPUT_BASE_DIR
-                value: /data/data_portal_output/test_data_portal
+                value: /data/raw_data/test_data_portal
               - name: RAW_DATA_PUBLISH_DIR
                 value: portal_publish_dir/test_data_portal
               - name: DATA_PORTAL_TYPE
@@ -104,8 +104,6 @@ spec:
                 value: "300000"
               - name: MERGE_BUFFER_SIZE
                 value: "4096"
-              - name: WRITE_BUFFER_SIZE
-                value: "10000000"
               - name: INPUT_DATA_FORMAT
                 value: TF_RECORD
               - name: COMPRESSED_TYPE

diff --git a/web_console/tests/utils/job_builder.test.js b/web_console/tests/utils/job_builder.test.js
@@ -243,7 +243,6 @@ describe('portalGenerateYaml', () => {
         batch_size: 1024,
         max_flying_item: 300000,
         merge_buffer_size: 4096,
-        write_buffer_size: 10000000,
         input_data_format: 'TF_RECORD',
         output_data_format: 'TF_RECORD',
         compressed_type: '',

diff --git a/web_console/utils/job_builder.js b/web_console/utils/job_builder.js
@@ -260,7 +260,6 @@ function portalGenerateYaml(federation, raw_data) {
             { name: 'BATCH_SIZE', value: String(raw_data.context.batch_size) },
             { name: 'MAX_FLYING_ITEM', value: String(raw_data.context.max_flying_item) },
             { name: 'MERGE_BUFFER_SIZE', value: String(raw_data.context.merge_buffer_size) },
-            { name: 'WRITE_BUFFER_SIZE', value: String(raw_data.context.write_buffer_size) },
             { name: 'INPUT_DATA_FORMAT', value: raw_data.context.input_data_format },
             { name: 'COMPRESSED_TYPE', value: raw_data.context.compressed_type },
             { name: 'OUTPUT_DATA_FORMAT', value: raw_data.context.output_data_format },