Add option for FanOut on CSV Extract, readd schema init on BigQueryLoad

Author: qqndrew

API/pom.xml

@@ -31,7 +31,7 @@
     <parent>
         <groupId>org.ohnlp.backbone</groupId>
         <artifactId>backbone-parent</artifactId>
-        <version>3.0.21</version>
+        <version>3.0.22</version>
     </parent>
 
     <artifactId>api</artifactId>

Core/pom.xml

@@ -7,7 +7,7 @@
     <parent>
         <groupId>org.ohnlp.backbone</groupId>
         <artifactId>backbone-parent</artifactId>
-        <version>3.0.21</version>
+        <version>3.0.22</version>
     </parent>
 
     <artifactId>core</artifactId>

Example-Backbone-Configs/pom.xml

@@ -7,7 +7,7 @@
     <parent>
         <groupId>org.ohnlp.backbone</groupId>
         <artifactId>backbone-parent</artifactId>
-        <version>3.0.21</version>
+        <version>3.0.22</version>
     </parent>
 
     <artifactId>example-backbone-configs</artifactId>

IO/pom.xml

@@ -7,7 +7,7 @@
     <parent>
         <groupId>org.ohnlp.backbone</groupId>
         <artifactId>backbone-parent</artifactId>
-        <version>3.0.21</version>
+        <version>3.0.22</version>
     </parent>
 
     <groupId>org.ohnlp.backbone.io</groupId>

IO/src/main/java/org/ohnlp/backbone/io/bigquery/BigQueryLoad.java

@@ -34,6 +34,7 @@ public void init() throws ComponentInitializationException {
 
     @Override
     public POutput expand(PCollection<Row> input) {
+        this.writeSchema = input.getSchema();
         return input.apply(
                 "Transform output rows to BigQuery TableRow format", ParDo.of(
                         new DoFn<Row, TableRow>() {

IO/src/main/java/org/ohnlp/backbone/io/local/CSVExtract.java

@@ -52,6 +52,13 @@ public class CSVExtract extends ExtractToMany {
     )
     private boolean skipFirstRow;
 
+    @ConfigurationProperty(
+            path = "fanout",
+            desc = "Whether to reshuffle records after read. Defaults to false. Set to true if fanout is needed (i.e., if your individual CSV files are very large)",
+            required = false
+    )
+    private boolean fanout = false;
+
     @ConfigurationProperty(
             path = "schema",
             desc = "CSV File Schema"
@@ -65,11 +72,11 @@ public class CSVExtract extends ExtractToMany {
 
     @Override
     public void init() throws ComponentInitializationException {
-
     }
 
     @Override
     public PCollectionRowTuple expand(PBegin input) {
+
         PCollection<String> fileURIs = input.apply("Scan Input Directory for Partitioned Files", Create.of(Arrays.stream(Objects.requireNonNull(new File(dir).listFiles())).map(f -> f.toURI().toString()).collect(Collectors.toList()))).setCoder(StringUtf8Coder.of());
         PCollectionTuple readColls = fileURIs.apply("Read CSV Records and Map to Rows", ParDo.of(new DoFn<String, Row>() {
             private String[] header;
@@ -127,7 +134,10 @@ public void processFile(ProcessContext pc) {
         }).withOutputTags(new TupleTag<>("CSV Records"), TupleTagList.of(new TupleTag<>("Errored Records"))));
         PCollection<Row> read = readColls.get("CSV Records");
         read.setRowSchema(this.schema);
-        PCollectionRowTuple ret = PCollectionRowTuple.of("CSV Records", read.apply("Break Fusion", Repartition.of())).and("Errored Records", readColls.get("Errored Records"));
+        if (this.fanout) {
+            read = read.apply("Break Fusion", Repartition.of());
+        }
+        PCollectionRowTuple ret = PCollectionRowTuple.of("CSV Records", read).and("Errored Records", readColls.get("Errored Records"));
         // Set Coders
         ret.get("CSV Records").setRowSchema(this.schema);
         ret.get("Errored Records").setRowSchema(this.errorSchema);

Plugin-Manager/pom.xml

@@ -7,7 +7,7 @@
     <parent>
         <groupId>org.ohnlp.backbone</groupId>
         <artifactId>backbone-parent</artifactId>
-        <version>3.0.21</version>
+        <version>3.0.22</version>
     </parent>
 
     <artifactId>plugin-manager</artifactId>

Transforms/pom.xml

@@ -7,7 +7,7 @@
     <parent>
         <groupId>org.ohnlp.backbone</groupId>
         <artifactId>backbone-parent</artifactId>
-        <version>3.0.21</version>
+        <version>3.0.22</version>
     </parent>
 
     <groupId>org.ohnlp.backbone.transforms</groupId>

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>org.ohnlp.backbone</groupId>
     <artifactId>backbone-parent</artifactId>
-    <version>3.0.21</version>
+    <version>3.0.22</version>
 
 
     <properties>