public void bulkImportRcvToFluo()

in modules/spark/src/main/java/org/apache/fluo/recipes/spark/FluoSparkHelper.java [135:151]

• 14 lines of code
• 2 McCabe index (conditional complexity)

  public void bulkImportRcvToFluo(JavaPairRDD<RowColumn, Bytes> data, BulkImportOptions opts) {

    data = partitionForAccumulo(data, fluoConfig.getAccumuloTable(), opts);

    JavaPairRDD<Key, Value> kvData = data.flatMapToPair(tuple -> {
      List<Tuple2<Key, Value>> output = new LinkedList<>();
      RowColumn rc = tuple._1();
      FluoKeyValueGenerator fkvg = new FluoKeyValueGenerator();
      fkvg.setRow(rc.getRow()).setColumn(rc.getColumn()).setValue(tuple._2().toArray());
      for (FluoKeyValue kv : fkvg.getKeyValues()) {
        output.add(new Tuple2<>(kv.getKey(), kv.getValue()));
      }
      return output.iterator();
    });

    bulkImportKvToAccumulo(kvData, fluoConfig.getAccumuloTable(), opts);
  }