Sunil's Notes: February 2016

Difference between reduce() and fold() method on Spark RDD

When you can call fold() method on the RDD it returns a different result than you normally expect, so i wanted to figure out how fold() method actually works so i built this simple application

	package com.spnotes.spark

	import org.apache.spark.{Logging, SparkContext, SparkConf, Partitioner}

	/**
	* Created by sunilpatil on 2/16/16.
	*/
	object HelloSparkPartitions extends Logging{

	def main(argv:Array[String]): Unit ={
	val sparkConf = new SparkConf().setMaster("local[1]").setAppName("HelloSparkPartitions")
	val sparkContext = new SparkContext(sparkConf)

	val input = sparkContext.parallelize(List(1,2,3,4,5,6,7,8),3)
	println("******************** mapPartitions *****************")
	input.mapPartitions(mapPartition).collect()
	println("******************** reduce *****************")
	println( "input.reduce " + input.reduce((x,y)=> add(x,y)))
	println("******************** fold *****************")
	println( "input.fold " + input.fold(10)((x,y)=> add(x,y)))
	}

	def add(x:Int, y:Int): Int={
	logError(s"Inside add -> $x, $y")
	x +y
	}

	def mapPartition(iterator: Iterator[Int]): Iterator[Int] ={
	logDebug("Inside mapPartition ")
	iterator.foreach(println)
	iterator
	}
	}

view raw HelloSparkPartitions.scala hosted with ❤ by GitHub

First thing that i do in the application is create a simple RDD with 8 values from 1 to 8 and divide it into 3 partitions sparkContext.parallelize(List(1,2,3,4,5,6,7,8),3). Then i am calling input.mapPartitions(mapPartition) to iterate through all the partitions in the RDD and printing records in them one by one. This shows that the RDD has 3 partitions and 1 and 2 are in first partitions 3,4,5 are in second partions and record 6,7,8 are in the third partitions. Then next step is to call input.reduce((x,y)=> add(x,y))) method that will invoke add reduce function on the RDD, as you can see the output. The reduce function simply starts calling add method first for first 2 records then it starts calling it with running count for rest of the elements in the RDD The last part is fold() method which i am calling with initial value of 10. As you can see from the output of fold() method, it first takes 10 as initial value and adds all the elements in single partitions to it. But then it also takes running counts across the RDDs adds 10 to it sums them up. Because of this, the result of fold() = (initial value * num of partitions +1) + sum of reduce


********************** mapPartitions *******************
[Stage 0:>(0 + 0) / 3]2016-02-17 10:22:13 DEBUG HelloSparkPartitions:63 - Inside mapPartition 
1
2
2016-02-17 10:22:13 DEBUG HelloSparkPartitions:63 - Inside mapPartition 
3
4
5
2016-02-17 10:22:13 DEBUG HelloSparkPartitions:63 - Inside mapPartition 
6
7
8
********************** reduce *******************
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 1, 2
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 3, 4
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 7, 5
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 3, 12
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 6, 7
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 13, 8
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 15, 21
input.reduce 36
********************** fold *******************
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 10, 1
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 11, 2
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 10, 13
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 10, 3
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 13, 4
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 17, 5
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 23, 22
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 10, 6
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 16, 7
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 23, 8
2016-02-17 10:22:13 ERROR HelloSparkPartitions:75 - Inside add -> 45, 31
input.fold 76

How to parse fillable PDF form in Java

I wanted to figure out how to parse a fillable PDF form in Java, so that i could do some processing on it. So i built this sample PDFFormParsingPOC project that uses Apache PDFBox library. This is simple java class that i built, in which i read the PDF file first and then parse it into PDDocument. Then i can get all the fields in the PDF form by calling PDDocument.getDocumentCatalog().getAcroForm().getFields() and start iterating through it. For every field that i find, first i try to figure out what is the type of the field and then use it to print the field with its name and value to console

	package com.sptech.pdfparsed;

	import org.apache.pdfbox.pdmodel.PDDocument;
	import org.apache.pdfbox.pdmodel.PDDocumentCatalog;
	import org.apache.pdfbox.pdmodel.common.PDMetadata;
	import org.apache.pdfbox.pdmodel.interactive.form.PDAcroForm;
	import org.apache.pdfbox.pdmodel.interactive.form.PDCheckbox;
	import org.apache.pdfbox.pdmodel.interactive.form.PDChoiceField;
	import org.apache.pdfbox.pdmodel.interactive.form.PDTextbox;

	import java.io.File;
	import java.util.List;

	public class HelloPDFBox {
	public static void listFields(PDDocument doc) throws Exception {

	PDDocumentCatalog catalog = doc.getDocumentCatalog();
	PDMetadata pdMetadata = catalog.getMetadata();
	PDAcroForm form = catalog.getAcroForm();
	List fields = form.getFields();

	System.out.println("Fields " + fields.get(0).getClass());

	for (Object field : fields) {

	if (field instanceof PDTextbox) {
	PDTextbox pdTextbox = (PDTextbox) field;
	System.out.println("PDTextBox " + pdTextbox.getFullyQualifiedName() + " " + pdTextbox.getValue());
	} else if (field instanceof PDChoiceField) {
	PDChoiceField pdChoiceField = (PDChoiceField) field;
	System.out.println("PDChoice Field " + pdChoiceField.getFullyQualifiedName() + " " + pdChoiceField.getValue());
	} else if (field instanceof PDCheckbox) {
	PDCheckbox pdCheckbox = (PDCheckbox) field;
	System.out.println("PDCheckbox Field " + pdCheckbox.getFullyQualifiedName() + " " + pdCheckbox.getValue());
	} else {
	System.out.print(field);
	System.out.print(" = ");
	System.out.print(field.getClass());
	System.out.println();
	}
	}

	}

	public static void main(String[] args) throws Exception {
	if (args.length != 1) {
	System.out.println("Please specify fully qualified path of the PDF form");
	System.exit(-1);
	}
	String fileName = args[0];
	System.out.println("Parsing PDF Form at " + fileName);
	File file = new File(fileName);
	PDDocument doc = PDDocument.load(file);
	listFields(doc);
	}

	}

view raw HelloPDFBox.java hosted with ❤ by GitHub

You can download the Apache PDFBox project and execute it by passing fully qualified name of the fillable PDF form and it will print out field name value pairs to console. If you dont have a pdf form already you can download Sample Fillable PDF Form

Invoking Python from Spark Scala project

When your developing your Spark code, you have option of developing it using either Scala, Java or Python. In some cases you might want to mix the languages that you want to use. I wanted to try that out so i built this simple Spark program that passes control to Python for performing transformation (All that it does it append word "python " in front of every line). You can download source code for sample project from here First thing that i did was to develop this simple python script that reads one line at a time from console, appends "Python " to the line and writes it back to standard console

	#!/usr/bin/env python

	import sys

	for line in sys.stdin:
	sys.stdout.write("Python "+line)

view raw echo.py hosted with ❤ by GitHub

Now this is how the driver looks like, most of the spark code is same only difference is lines.pipe("python echo.py") which says that pass every line in the RDD to python echo.py. and collect the output. Now there is nothing specific to python here, instead you could use any executable.

	package com.spnotes

	import org.apache.spark.{SparkContext, SparkConf}

	object ScalaPython {

	def main(argv:Array[String]): Unit ={
	System.out.println("Entering ScalaPython.main")
	if(argv.length != 1){
	println("Please provide 1 parameters <inputFileName>")
	System.exit(1)
	}
	val fileName = argv(0)
	val sparkConf = new SparkConf().setAppName("ScalaPython").setMaster("local")
	val sparkContext = new SparkContext(sparkConf)

	val lines = sparkContext.textFile(fileName)
	println("Print lines from original file")
	lines.foreach(println)
	val lines1 = lines.pipe("python echo.py")
	println("Print lines returned from python")

	lines1.foreach(println)
	System.out.println("Exiting ScalaPython.main")

	}

	}

view raw ScalaPython.scala hosted with ❤ by GitHub

When you run this code in cluster you should copy the python file on your machine say in spark directory then you can execute


bin/spark-submit 
    --files echo.py  
    ScalaPython-1.0-SNAPSHOT-jar-with-dependencies.jar helloworld.txt

How to use HBase sink with Flume

I wanted to figure out how to use HBase as target for flume, so i created this sample configuration which reads events from netcat and writes them to HBase.

First step is to create test table in HBase with CF1 as column family. Everytime Flume gets a event it will write to HBase in test table in CF1 column family
```
create 'test','CF1'
```

Create Flume configuration file that looks like this, I am using HBase sink with SimpleHbaseEventSerializer as Event Serializer. Note that i am assuming that this is unsecured cluster (Sandbox), but if you have secured cluster you should follow steps mentioned in Configure a Secure HBase Sink

	agent1.sources = netcat1
	agent1.sinks = hbase
	agent1.channels =memory1

	agent1.sources.netcat1.type = netcat
	agent1.sources.netcat1.bind = localhost
	agent1.sources.netcat1.port = 44444

	agent1.sinks.hbase.type=hbase
	agent1.sinks.hbase.table=test
	agent1.sinks.hbase.columnFamily=CF1
	agent1.sinks.hbase.serializer=org.apache.flume.sink.hbase.SimpleHbaseEventSerializer

	agent1.channels.memory1.type = memory
	agent1.channels.memory1.capacity = 1000
	agent1.channels.memory1.transactionCapacity = 100

	agent1.sources.netcat1.channels = memory1
	agent1.sinks.hbase.channel=memory1

view raw netcat-hbase.properties hosted with ❤ by GitHub

Start the Flume server with the following command


bin/flume-ng agent --conf conf --conf-file conf/netcat-hbase.properties --name agent1 -Dflume.root.logger=DEBUG,console

Now open the netcat client on port 44444 and send some messages to flume
If you query HBase test table, you should see the messages that were published to netcat