### Question:
CommonCrawlIndex { public static FlowDef buildFlowDef(Properties properties){ Tap source=null; Boolean isDistributed =((properties.containsKey("platform")) && properties.getProperty("platform").toString().compareTo("DISTRIBUTED")==0); String inPath = properties.getProperty("inPath"); if (isDistributed){ source = new Hfs(new cascading.scheme.hadoop.TextLine(new Fields("line")), inPath); }else { source = new FileTap(new cascading.scheme.local.TextLine(new Fields("line")), inPath); } Tap sink = new EsTap(properties.getProperty("es.target.index")); return CommonCrawlIndex.createCommonCrawlFlowDef(source, sink); }  static FlowDef buildFlowDef(Properties properties); static FlowDef createCommonCrawlFlowDef(Tap source, Tap sink);  }### Answer:
@Test public void testMain() throws IOException { Properties properties = new ConfigReader().renderProperties(CommonCrawlIndexTest.class); FlowDef flowDef = CommonCrawlIndex.buildFlowDef(properties); if (properties.getProperty("platform").toString().compareTo("LOCAL")==0){ new LocalFlowConnector(properties).connect(flowDef).complete(); } else { new HadoopFlowConnector(properties).connect(flowDef).complete(); } }