问题:使用MultiSinkTap生成多个输出文件

我有以下数据集作为输入

id,name,gender
asinha161,Aniruddha,Male
vic,Victor,Male
day1,Daisy,Female
jazz030,Jasmine,Female
Mic002,Michael,Male

我的目标是将男性和女性分成两个单独的输出文件,如下所示

男性数据集

id,name,gender
asinha161,Aniruddha,Male
vic,Victor,Male
Mic002,Michael,Male

女性数据集

id,name,gender
day1,Daisy,Female
jazz030,Jasmine,Female

现在,我尝试编写一个应该执行上述任务的级联框架代码,代码如下

public class Main {

      public static void main(String[] args) {
          Tap sourceTap = new FileTap(new TextDelimited(true, ","),       "inputFile.txt");
          Tap sink_one = new FileTap(new TextDelimited(true, ","), "maleFile.txt");
          Tap sink_two = new FileTap(new TextDelimited(true, ","), "FemaleFile.txt");

          Pipe assembly = new Pipe("inputPipe");


          // ...split into two pipes
          Pipe malePipe = new Pipe("for_male", assembly);
          malePipe=new Each(malePipe,new CustomFilterByGender("male"));
          Pipe femalePipe = new Pipe("for_female", assembly);
          femalePipe=new Each(femalePipe, new CustomFilterByGender("female"));
          // create the flow
           List<Pipe> pipes = new ArrayList<Pipe>(2)
        {{pipes.add(countOne);
          pipes.add(countTwo);}};

          Tap outputTap=new MultiSinkTap<>(sink_one,sink_two);

          FlowConnector flowConnector = new LocalFlowConnector();
          Flow flow = flowConnector.connect(sourceTap, outputTap, pipes);
          flow.complete();
      }

其中CustomFilterByGender(字符串性别);是一个自定义函数,它根据作为参数传递的性别值返回元组。

请注意,为了效率,我没有使用自定义缓冲区。

使用 MultiSinkTap,我无法获得所需的输出,因为LocalFlowConnector对象的connect()方法不接受导致编译时错误的 MultiSinkTap 对象。

如果您建议对上述代码进行可能的更改以使其工作或使用MultiSinkTap的方式,这将是必不可少的。

感谢您耐心地回答问题:)

解答

我认为您想将不同管道的输出写入不同的输出文件,我对您的代码进行了一些更改,这肯定可以解决您的目的。

public class Main {
  public static void main(String[] args) {
      Tap sourceTap = new FileTap(new TextDelimited(true, ","), "inputFile.txt");
      Tap sink_one = new FileTap(new TextDelimited(true, ","), "maleFile.txt");
      Tap sink_two = new FileTap(new TextDelimited(true, ","), "FemaleFile.txt");

      Pipe assembly = new Pipe("inputPipe");

      Pipe malePipe = new Pipe("for_male", assembly);
      malePipe=new Each(malePipe,new CustomFilterByGender("male"));
      Pipe femalePipe = new Pipe("for_female", assembly);
      femalePipe=new Each(femalePipe, new CustomFilterByGender("female"));

      List<Pipe> pipes = new ArrayList<Pipe>(2);
      pipes.add(malePipe);
      pipes.add(femalePipe);

      Map<String, Tap> sinks = new HashMap<String, Tap>();
      sinks.put("for_male", sink_one);
      sinks.put("for_female", sink_two);

      FlowConnector flowConnector = new LocalFlowConnector();
      Flow flow = flowConnector.connect(sourceTap, sinks, pipes);
      flow.complete();
  }

除了使用 MultiSinkTap 之外,您可以直接将您想要连接到输出管道的 Sink 的 Map<> 在这种情况下提供为 malePipe 和 femalePipe。