Merge pull request #2 from navinrathore/pipeUtilProps

Pipe's props is initialized with blank hashmap. makes getProps() never return null

Author: navinrathore

client/src/main/java/zingg/client/pipe/Pipe.java

@@ -32,7 +32,7 @@ public class Pipe implements Serializable{
 	String name;
 	Format format;
 	String preprocessors;
-	Map<String, String> props;
+	Map<String, String> props = new HashMap<String, String>();
 	@JsonSerialize(using = CustomSchemaSerializer.class)
 	StructType schema = null;
 	Map<String, String> sparkProps;

core/src/main/java/zingg/Labeller.java

@@ -199,7 +199,7 @@ protected void printMarkedRecordsStat() {
 		System.out.println(msg);
 	}
 
-	protected void writeLabelledOutput(Dataset<Row> records) {
+	protected void writeLabelledOutput(Dataset<Row> records) throws ZinggClientException {
 		if (records == null) {
 			LOG.warn("No records to be labelled.");
 			return;

core/src/main/java/zingg/Linker.java

@@ -50,7 +50,7 @@ protected Dataset<Row> selectColsFromBlocked(Dataset<Row> blocked) {
 		return blocked;
 	}
 
-	public void writeOutput(Dataset<Row> sampleOrginal, Dataset<Row> dupes) {
+	public void writeOutput(Dataset<Row> sampleOrginal, Dataset<Row> dupes) throws ZinggClientException {
 		try {
 			// input dupes are pairs
 			/// pick ones according to the threshold by user

core/src/main/java/zingg/Matcher.java

@@ -160,7 +160,7 @@ public void execute() throws ZinggClientException {
 		}
     }
 
-	public void writeOutput(Dataset<Row> blocked, Dataset<Row> dupesActual) {
+	public void writeOutput(Dataset<Row> blocked, Dataset<Row> dupesActual) throws ZinggClientException {
 		try{
 		//input dupes are pairs
 		///pick ones according to the threshold by user

core/src/main/java/zingg/TrainingDataFinder.java

@@ -139,7 +139,7 @@ public void execute() throws ZinggClientException {
 			}	
     }
 
-	public void writeUncertain(Dataset<Row> dupesActual, Dataset<Row> sampleOrginal) {
+	public void writeUncertain(Dataset<Row> dupesActual, Dataset<Row> sampleOrginal) throws ZinggClientException {
 		//dupesActual.show(4);
 		//input dupes are pairs
 		dupesActual = DFUtil.addClusterRowNumber(dupesActual, spark);

core/src/main/java/zingg/util/BlockingTreeUtil.java

@@ -77,7 +77,7 @@ public static Tree<Canopy> createBlockingTreeFromSample(Dataset<Row> testData,
 		return createBlockingTree(sample, positives, sampleFraction, blockSize, args, hashFunctions);
 	}
 
-	public static void writeBlockingTree(SparkSession spark, JavaSparkContext ctx, Tree<Canopy> blockingTree, Arguments args) throws Exception {
+	public static void writeBlockingTree(SparkSession spark, JavaSparkContext ctx, Tree<Canopy> blockingTree, Arguments args) throws Exception, ZinggClientException {
 		byte[] byteArray  = Util.convertObjectIntoByteArray(blockingTree);
 		StructType schema = DataTypes.createStructType(new StructField[] { DataTypes.createStructField("BlockingTree", DataTypes.BinaryType, false) });
 		List<Object> objList = new ArrayList<>();

core/src/main/java/zingg/util/PipeUtil.java

@@ -192,7 +192,8 @@ public static Dataset<Row> read(SparkSession spark, boolean addLineNo, int numPa
 		return rows;
 	}
 
-	public static void write(Dataset<Row> toWriteOrig, Arguments args, JavaSparkContext ctx, Pipe... pipes) {
+	public static void write(Dataset<Row> toWriteOrig, Arguments args, JavaSparkContext ctx, Pipe... pipes) throws ZinggClientException {
+		try {
 			for (Pipe p: pipes) {
 			Dataset<Row> toWrite = toWriteOrig;
 			DataFrameWriter writer = toWrite.write();
@@ -293,11 +294,13 @@ else if (p.getFormat().equals(Format.JDBC)){
 			}
 
 
+			}
+		} catch (Exception ex) {
+			throw new ZinggClientException(ex.getMessage());
 		}
-
 	}
 
-	public static void writePerSource(Dataset<Row> toWrite, Arguments args, JavaSparkContext ctx, Pipe[] pipes ) {
+	public static void writePerSource(Dataset<Row> toWrite, Arguments args, JavaSparkContext ctx, Pipe[] pipes ) throws ZinggClientException {
 		List<Row> sources = toWrite.select(ColName.SOURCE_COL).distinct().collectAsList();
 		for (Row r : sources) {
 			Dataset<Row> toWriteNow = toWrite.filter(toWrite.col(ColName.SOURCE_COL).equalTo(r.get(0)));