stub reduce_bucket

diff --git a/scripts/paraexport b/scripts/paraexport
index 903569e64e34e0ba1e3481c51e934f2cf01e8610..bb8c316b94d7deff11c2e5a8eea5a87452b283b3 100755 (executable)
--- a/scripts/paraexport
+++ b/scripts/paraexport
@@ -1,10 +1,26 @@
 #!/bin/sh
 
 # start multiple jobs in parallel
-target/debug/polarsbop mapbucket 0 &
-target/debug/polarsbop mapbucket 1 &
-target/debug/polarsbop mapbucket 2 &
+# target/debug/polarsbop mapbucket 0 &
+# target/debug/polarsbop mapbucket 1 &
+# target/debug/polarsbop mapbucket 2 &
 
-wait %1 %2 %3
+# wait %1 %2 %3
 
-echo done
+# echo mapbucket stage done
+
+target/debug/polarsbop reducebucket 0
+
+# target/debug/polarsbop reducebucket 0 &
+# target/debug/polarsbop reducebucket 1 &
+# target/debug/polarsbop reducebucket 2 &
+# target/debug/polarsbop reducebucket 3 &
+# target/debug/polarsbop reducebucket 4 &
+# target/debug/polarsbop reducebucket 5 &
+# target/debug/polarsbop reducebucket 6 &
+# target/debug/polarsbop reducebucket 7 &
+# target/debug/polarsbop reducebucket 8 &
+
+# wait %4
+
+echo reducebucket stage done

diff --git a/src/main.rs b/src/main.rs
index 93bcac406571c0dec9e718803092587546a0369b..788a01fe7e204858ef37b5fdc9de9bd52b149938 100644 (file)
--- a/src/main.rs
+++ b/src/main.rs
@@ -82,12 +82,12 @@ async fn map_bucket(partition: &str) -> Result<(),PolarsError> {
 
        // prep output file
        let bucketdir = format!("bucket-{}",bindex);
-       create_dir_all(format!("out/{}",bucketdir.clone()))
+       create_dir_all(format!("out/mapbucket/{}",bucketdir.clone()))
            .expect("couldn't create out dir");
 
        let outfilename = format!("partition-{}.json",partition);
-       let outs3key = format!("{}/{}",bucketdir,outfilename);
-       let outfilepath = format!("out/{}/{}",bucketdir,outfilename);
+       let outs3key = format!("mapbucket/{}/{}",bucketdir,outfilename);
+       let outfilepath = format!("out/mapbucket/{}/{}",bucketdir,outfilename);
 
        // println!("{}",outfilepath);
        let mut file = File::create(outfilepath.clone())
@@ -127,6 +127,10 @@ async fn map_bucket(partition: &str) -> Result<(),PolarsError> {
     Ok(())
 }
 
+async fn reduce_bucket(bucket: &str) -> Result<(),PolarsError> {
+    Ok(())
+}
+
 #[::tokio::main]
 async fn main() -> Result<(),PolarsError> {
 
@@ -144,6 +148,13 @@ async fn main() -> Result<(),PolarsError> {
 
            map_bucket(partition.as_str()).await
        },
+       "reducebucket" => {
+           // expect next arg to be a bucket partition = 0-8
+           let bucket = argsiter.next()
+               .expect("single argument required, 0 - 8");
+
+           reduce_bucket(bucket.as_str()).await
+       },
        _ => {
            panic!("invalid command");
        }